Introduction to MQTT
SIM900
SIM900 enables GPRS for embedded applications. We can implement MQTT (Message Queue Telemetry Transport) Client protocol using SIM900 TCP function AT Commands.
MQTT is a lightweight publish-subscribe-based messaging protocol.
It is quicker (faster) than other request-response based APIs like HTTP.
It is developed on the base of the TCP/IP protocol.
It allows remote location devices to connect, subscribe, publish, etc. to a specific topic on the server with the help of a message broker.
MQTT Broker/Message broker is a module in between the sender and the receiver. It is an element for message validation, transformation, and routing.
The broker is responsible for distributing messages to the interested clients (subscribed clients) of their interested topic.
For example, if the temperature sensor publishes the temperature data (message) on the topic “temperature” then interested clients who have subscribed to the “temperature” topic get that published temperature data as shown in the above figure.
MQTT is widely used in IoT (Internet of Things) embedded applications, where every sensor is connected to a server and we have access to control them over the internet.
To know about SIM900 GSM/GPRS Module refer to SIM900
Connection Diagram of SIM900 GSM Module with PIC18F4550
PIC18F4550 Interface with SIM900
MQTT Client over GPRS
Let’s program PIC18F4550 to configure SIM900A as MQTT Client and Subscribe/Publish data from/to Server using GPRS.
Here we are using the Adafruit server for MQTT Client demo purpose.
In the IOT platform, Adafruit IO Dashboard allows us to visualize and provides control over the connected device to the internet. Anyone can visualize and analyze live data from their sensor devices. To learn more and start with Adafruit IO Dashboard refer link
Just sign up and create a dashboard. After the successful creating of the dashboard, we will get the AIO key which is later used to access feed data.
Example
Now let’s program PIC18F4550 to control LED brightness and monitor POT status on a remote location from the Adafruit dashboard.
Once we created a dashboard on Adafruit we can add various blocks that can be used to control devices as well as monitor the status of devices. To see more about blocks, refer link
In the below program of MQTT Client, do the following
For MQTT Client Subscribe Demo
#define SUBSRCIBE_DEMO /* Define SUBSRCIBE demo */
//#define PUBLISH_DEMO /* Define PUBLISH demo */For MQTT Client Publish Demo
//#define SUBSRCIBE_DEMO /* Define SUBSRCIBE demo */
#define PUBLISH_DEMO /* Define PUBLISH demo */Edit Fields below with respective data
/* Define Required fields shown below */
#define AIO_SERVER "io.adafruit.com" /* Adafruit server */
#define AIO_SERVER_PORT "1883" /* Server port */
#define AIO_BASE_URL "/api/v2" /* Base URL for api */
#define AIO_USERNAME "Enter Username" /* Enter username here */
#define AIO_KEY "Enter AIO key" /* Enter AIO key here */
#define AIO_FEED "Enter Feed Key" /* Enter feed key */
#define APN "internet" /* APN of n/w service provider */
#define USERNAME ""
#define PASSWORD ""
In the below program, we are using response-based functions to get the better status if things deviate from normal.
MQTT Packet Formation
MQTT uses many packet formats that used to connect to the server and subscribe or publish to the topic on the server.
Refer below link for MQTT OASIS standard. It will help to understand MQTT packet formations.
MQTT Client Code for PIC18F4550
/*
* PIC_GPRS_MQTTClient
* http://www.electronicwings.com
*
*/
#include "SIM900TCPClient.h" /* Include TCP Client header file */
#include "math.h"
#define MQTT_PROTOCOL_LEVEL 4
#define MQTT_CTRL_CONNECT 0x1
#define MQTT_CTRL_CONNECTACK 0x2
#define MQTT_CTRL_PUBLISH 0x3
#define MQTT_CTRL_PUBACK 0x4
#define MQTT_CTRL_PUBREC 0x5
#define MQTT_CTRL_PUBREL 0x6
#define MQTT_CTRL_PUBCOMP 0x7
#define MQTT_CTRL_SUBSCRIBE 0x8
#define MQTT_CTRL_SUBACK 0x9
#define MQTT_CTRL_UNSUBSCRIBE 0xA
#define MQTT_CTRL_UNSUBACK 0xB
#define MQTT_CTRL_PINGREQ 0xC
#define MQTT_CTRL_PINGRESP 0xD
#define MQTT_CTRL_DISCONNECT 0xE
#define MQTT_QOS_1 0x1
#define MQTT_QOS_0 0x0
/* Adjust as necessary, in seconds */
#define MQTT_CONN_KEEPALIVE 60
#define MQTT_CONN_USERNAMEFLAG 0x80
#define MQTT_CONN_PASSWORDFLAG 0x40
#define MQTT_CONN_WILLRETAIN 0x20
#define MQTT_CONN_WILLQOS_1 0x08
#define MQTT_CONN_WILLQOS_2 0x18
#define MQTT_CONN_WILLFLAG 0x04
#define MQTT_CONN_CLEANSESSION 0x02
#define DEFAULT_BUFFER_SIZE 200
#define DEFAULT_TIMEOUT 10000
#define DEFAULT_CRLF_COUNT 2
#define MINTHR 8000
#define RESOLUTION 488
#define InternalOsc_8MHz 8000000
#define InternalOsc_4MHz 4000000
#define InternalOsc_2MHz 2000000
#define InternalOsc_1MHz 1000000
#define InternalOsc_500KHz 500000
#define InternalOsc_250KHz 250000
#define InternalOsc_125KHz 125000
#define InternalOsc_31KHz 31000
#define Timer2Prescale_1 1
#define Timer2Prescale_4 4
#define Timer2Prescale_16 16
/* Select Demo */
#define SUBSRCIBE_DEMO /* Define SUBSRCIBE demo */
//#define PUBLISH_DEMO /* Define PUBLISH demo */
#define AIO_SERVER "io.adafruit.com" /* Adafruit server */
#define AIO_SERVER_PORT "1883" /* Server port */
#define AIO_BASE_URL "/api/v2" /* Base URL for api */
#define AIO_USERNAME "Enter Username" /* Enter username here */
#define AIO_KEY "Enter AIO key" /* Enter AIO key here */
#define AIO_FEED "Enter Feed Key" /* Enter feed key */
#define APN "internet"
#define USERNAME ""
#define PASSWORD ""
int16_t packet_id_counter = 0;
char clientID[] ="";
char will_topic[] = "";
char will_payload[] ="";
uint8_t will_qos = 1;
uint8_t will_retain = 0;
uint16_t StringToUint16(uint8_t* _String)
{
uint16_t value = 0;
while ('0' == *_String || *_String == ' ' || *_String == '"')
{
_String++;
}
while ('0' <= *_String && *_String <= '9')
{
value *= 10;
value += *_String - '0';
_String++;
}
return value;
}
uint8_t* AddStringToBuf(uint8_t *_buf, const char *_string)
{
uint16_t _length = strlen(_string);
_buf[0] = _length >> 8;
_buf[1] = _length & 0xFF;
_buf+=2;
strncpy((char *)_buf, _string, _length);
return _buf + _length;
}
bool sendPacket(uint8_t *buffer, uint16_t len)
{
uint16_t sendlen;
if (TCPClient_connected())
{
sendlen = min(len, 250);
TCPClient_Send((char*)buffer, sendlen);
return true;
}
else
return false;
}
uint16_t readPacket(uint8_t *buffer, int16_t _timeout)
{
uint16_t len = 0;
while (TCPClient_DataAvailable() > 0 && _timeout >=0)
{
buffer[len++] = TCPClient_DataRead();
_timeout-= 10; MSdelay(10);
}
buffer[len] = 0;
return len;
}
uint8_t MQTT_ConnectToServer()
{
return TCPClient_Start(1, AIO_SERVER, AIO_SERVER_PORT);
}
uint16_t MQTT_connectpacket(uint8_t* packet)
{
uint8_t*_packet = packet;
uint16_t _length;
_packet[0] = (MQTT_CTRL_CONNECT << 4);
_packet+=2;
_packet = AddStringToBuf(_packet, "MQTT");
_packet[0] = MQTT_PROTOCOL_LEVEL;
_packet++;
_packet[0] = MQTT_CONN_CLEANSESSION;
if (will_topic && strlen(will_topic) != 0) {
_packet[0] |= MQTT_CONN_WILLFLAG;
if(will_qos == 1)
_packet[0] |= MQTT_CONN_WILLQOS_1;
else if(will_qos == 2)
_packet[0] |= MQTT_CONN_WILLQOS_2;
if(will_retain == 1)
_packet[0] |= MQTT_CONN_WILLRETAIN;
}
if (strlen(AIO_USERNAME) != 0)
_packet[0] |= MQTT_CONN_USERNAMEFLAG;
if (strlen(AIO_KEY) != 0)
_packet[0] |= MQTT_CONN_PASSWORDFLAG;
_packet++;
_packet[0] = MQTT_CONN_KEEPALIVE >> 8;
_packet++;
_packet[0] = MQTT_CONN_KEEPALIVE & 0xFF;
_packet++;
if (strlen(clientID) != 0) {
_packet = AddStringToBuf(_packet, clientID);
} else {
_packet[0] = 0x0;
_packet++;
_packet[0] = 0x0;
_packet++;
}
if (will_topic && strlen(will_topic) != 0) {
_packet = AddStringToBuf(_packet, will_topic);
_packet = AddStringToBuf(_packet, will_payload);
}
if (strlen(AIO_USERNAME) != 0) {
_packet = AddStringToBuf(_packet, AIO_USERNAME);
}
if (strlen(AIO_KEY) != 0) {
_packet = AddStringToBuf(_packet, AIO_KEY);
}
_length = _packet - packet;
packet[1] = _length-2;
return _length;
}
uint16_t MQTT_publishPacket(uint8_t *packet, const char *topic, char *data, uint8_t qos)
{
uint8_t *_packet = packet;
uint16_t _length = 0;
uint16_t Datalen=strlen(data);
_length += 2;
_length += strlen(topic);
if(qos > 0) {
_length += 2;
}
_length += Datalen;
_packet[0] = MQTT_CTRL_PUBLISH << 4 | qos << 1;
_packet++;
do {
uint8_t encodedByte = _length % 128;
_length /= 128;
if ( _length > 0 ) {
encodedByte |= 0x80;
}
_packet[0] = encodedByte;
_packet++;
} while ( _length > 0 );
_packet = AddStringToBuf(_packet, topic);
if(qos > 0) {
_packet[0] = (packet_id_counter >> 8) & 0xFF;
_packet[1] = packet_id_counter & 0xFF;
_packet+=2;
packet_id_counter++;
}
memmove(_packet, data, Datalen);
_packet+= Datalen;
_length = _packet - packet;
return _length;
}
uint16_t MQTT_subscribePacket(uint8_t *packet, const char *topic, uint8_t qos)
{
uint8_t *_packet = packet;
uint16_t _length;
_packet[0] = MQTT_CTRL_SUBSCRIBE << 4 | MQTT_QOS_1 << 1;
_packet+=2;
_packet[0] = (packet_id_counter >> 8) & 0xFF;
_packet[1] = packet_id_counter & 0xFF;
_packet+=2;
packet_id_counter++;
_packet = AddStringToBuf(_packet, topic);
_packet[0] = qos;
_packet++;
_length = _packet - packet;
packet[1] = _length-2;
return _length;
}
void PWM_Init() /* Initialize PWM */
{
TRISCbits.TRISC1 = 0; /* Set CCP2 pin as output for PWM out */
CCP2CON = 0x0C; /* Set PWM mode */
}
int setPeriodTo(unsigned long FPWM)/* Set period */
{
int clockSelectBits, TimerPrescaleBits;
int TimerPrescaleValue;
float period;
unsigned long FOSC, _resolution = RESOLUTION;
if (FPWM < MINTHR) {TimerPrescaleBits = 2; TimerPrescaleValue = Timer2Prescale_16;}
else {TimerPrescaleBits = 0; TimerPrescaleValue = Timer2Prescale_1;}
if (FPWM > _resolution) {clockSelectBits = 7; FOSC = InternalOsc_8MHz;}
else if (FPWM > (_resolution >>= 1)) {clockSelectBits = 6; FOSC = InternalOsc_4MHz;}
else if (FPWM > (_resolution >>= 1)) {clockSelectBits = 5; FOSC = InternalOsc_2MHz;}
else if (FPWM > (_resolution >>= 1)) {clockSelectBits = 4; FOSC = InternalOsc_1MHz;}
else if (FPWM > (_resolution >>= 1)) {clockSelectBits = 3; FOSC = InternalOsc_500KHz;}
else if (FPWM > (_resolution >>= 1)) {clockSelectBits = 2; FOSC = InternalOsc_250KHz;}
else if (FPWM > (_resolution >>= 1)) {clockSelectBits = 1; FOSC = InternalOsc_125KHz;}
else {clockSelectBits = 0; FOSC = InternalOsc_31KHz;}
period = ((float)FOSC / (4.0 * (float)TimerPrescaleValue * (float)FPWM)) - 1.0;
period = round(period);
OSCCON = ((clockSelectBits & 0x07) << 4) | 0x02;
PR2 = (int)period;
T2CON = TimerPrescaleBits;
TMR2 = 0;
T2CONbits.TMR2ON = 1; /* Turn ON Timer2 */
return (int)period;
}
void SetDutyCycleTo(float Duty_cycle, int Period)/* Set Duty cycle for given period */
{
int PWM10BitValue;
PWM10BitValue = 4.0 * ((float)Period + 1.0) * (Duty_cycle/100.0);
CCPR2L = (PWM10BitValue >> 2);
CCP2CON = ((PWM10BitValue & 0x03) << 4) | 0x0C;
}
void ADC_Init() /* ADC initialization function */
{
TRISA = 0xFF; /* Set as input port */
ADCON1 = 0x07; /* Set AN0-7 as Analog & other as digital */
ADCON2 = 0x92; /* Right Justified, 4Tad and Fosc/32. */
ADRESH = 0; /* Flush ADC output Register */
ADRESL = 0;
}
int ADC_Read(char channel) /* ADC channel read function */
{
ADCON0 =(ADCON0 & 0b11000011)|((channel<<2) & 0b00111100); /* Channel is selected i.e (CHS3:CHS0) */
ADCON0 |= ((1<<ADON)|(1<<GO)); /* Enable ADC and start conversion */
while(ADCON0bits.GO_nDONE==1); /* wait for End of conversion i.e. Go/done'=0 conversion completed */
return((ADRESH*256) | (ADRESL));/* Return 10-bit ADC result */
}
int main()
{
char buf[4];
uint8_t _buffer[150];
uint16_t len;
#ifdef SUBSRCIBE_DEMO
int period;
long KeepAliveTime;
PWM_Init(); /* Initialize PWM */
period = setPeriodTo(10000);/* 10KHz PWM frequency */
#endif
#ifdef PUBLISH_DEMO
int ADC_Value;
ADC_Init();
#endif
OSCCON = 0x72; /* Set internal clock to 8MHz */
USART_Init(9600); /* Initiate USART with 9600 baud rate */
INTCONbits.GIE=1; /* enable Global Interrupt */
INTCONbits.PEIE=1; /* enable Peripheral Interrupt */
PIE1bits.RCIE=1; /* enable Receive Interrupt */
while(!SIM900_Start());
TCPClient_Shut();
TCPClient_ConnectionMode(0); /* 0 = Single; 1 = Multi */
TCPClient_ApplicationMode(0); /* 0 = Normal Mode; 1 = Transperant Mode */
AttachGPRS();
while(!(TCPClient_Connect(APN, USERNAME, PASSWORD)));
while(1)
{
MQTT_ConnectToServer();
len = MQTT_connectpacket(_buffer);
sendPacket(_buffer, len);
len = readPacket(_buffer, 1000);
#ifdef PUBLISH_DEMO
while(TCPClient_connected())
{
memset(_buffer, 0, 150);
memset(buf, 0, 4);
ADC_Value = (float)ADC_Read(0)/10.4;
sprintf(buf, "%d", ADC_Value); /* Read ADC channel 0 and publish it in range 0-100 */
len = MQTT_publishPacket(_buffer, "Nivya151/feeds/test", buf, 1);/* topic format: "username/feeds/aio_feed" e.g. "Nivya151/feeds/test" */
sendPacket(_buffer, len);
}
TCPClient_Close();
MSdelay(1000);
#endif
#ifdef SUBSRCIBE_DEMO
uint8_t valuePointer=0;
memset(_buffer, 0, 150);
len = MQTT_subscribePacket(_buffer, "Nivya151/feeds/test", 1);/* topic format: "username/feeds/aio_feed" e.g. "Nivya151/feeds/test" */
sendPacket(_buffer, len);
KeepAliveTime = (MQTT_CONN_KEEPALIVE * 1000L);
while(KeepAliveTime > 0) /* Read subscription packets till Alive time */
{
len = readPacket(_buffer, 1000);
for (uint16_t i=0; i<len;i++)
{
for (uint8_t k = 0; k < 4; k++)/* here 4 is aio_feed char length e.g. aio_feed "test" has length of 4 char */
buf[k] = _buffer[i + k];
if (strstr(buf, AIO_FEED) != 0)
{
valuePointer = i + 4;
SetDutyCycleTo(StringToUint16(_buffer + valuePointer), period);
i = len;
}
}
MSdelay(1);
KeepAliveTime--;
}
if(TCPClient_connected()) TCPClient_Close();
MSdelay(1000);
#endif
}
}
Comments