result of success : 1 to 0, max msg size: 127B to 16383B, fixed existing bugs

Fork of MQTTClient by wakwak koba

MQTTClient.cpp

Committer:
wakwak_koba
Date:
2016-01-04
Revision:
11:2a8c86bfdeca
Parent:
10:6fa55ad359f0

File content as of revision 11:2a8c86bfdeca:

/*
MQTTClient.cpp
Based on MQTTClient from http://ceit.uq.edu.au/content/mqttclient-mbed-version-20
A simple MQTT client for mbed, version 2.0
By Yilun FAN, @CEIT, @JAN 2011

Bug fixes and additions by Andrew Lindsay (andrew [at] thiseldo [dot] co [dot] uk)

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
*/

#include "MQTTClient.h"

/** Default Constructor
 */
MQTTClient::MQTTClient() {}

/** Default Destructor
 */
MQTTClient::~MQTTClient() {}

/** Alternative Constructor with parameters
 *
 * Allow object to be constructed with minimum parameters.
 *
 * @param server The IP address of the server to connect to
 * @param port   The TCP/IP port on the server to connect to
 * @param callback Callback function to handle subscription to topics
 */
MQTTClient::MQTTClient(IpAddr server, int port, void (*callback)(char*, char*)) {
    this->port = port;
    callback_server = callback;

    host.setPort(port);
    host.setIp(server);
    host.setName("localhost");

    this->userName = NULL;
    this->password = NULL;
    connected = false;
    sessionOpened = false;
    
    pTCPSocket = NULL;
    timer.start();
}

/** Alternative Constructor with parameters
 *
 * Allow object to be constructed with minimum parameters.
 *
 * @param server The IP address of the server to connect to
 * @param port   The TCP/IP port on the server to connect to
 * @param id The client name shown on MQTT server.
 * @param userName Pointer to username string, zero terminated. Null for not used 
 * @param password Pointer to password string, zero terminated. Null for not used 
 * @param callback Callback function to handle subscription to topics
 */
MQTTClient::MQTTClient(IpAddr server, int port, char *id, char *userName, char *password, void (*callback)(char*, char*)) {
    this->port = port;
    callback_server = callback;

    host.setPort(port);
    host.setIp(server);
    host.setName("localhost");

    this->userName = userName;
    this->password = password;
    clientId = id;
    connected = false;
    sessionOpened = false;
    
    pTCPSocket = NULL;
    timer.start();
}

/** Send a message of specified size
 *
 * @param msg  message string
 * @param size Size of the message to send
 * @returns value to indicate message sent successfully or not, -1 for error, 0 for success.
 */
int MQTTClient::send_data(const char* msg, int size) {
    writtable = false;

/*    printf("send_data length %d\n", size);
    for (int i = 0; i < size; i++) {
        printf("%x, ", msg[i]);
    }
    printf("\r\n"); */

    int transLen = pTCPSocket->send(msg, size);
    pool();
    
    /*Check send length matches the message length*/
    if (transLen != size) {
        for (int i = 0; i < size; i++) {
            printf("%x, ", msg[i]);
        }
        printf("Error on send.\r\n");
        return -1;
    }
    
    return 0;
}

/** Start a MQTT session, build CONNECT packet
 *
 * @param id The client name shown on MQTT server.
 * @returns -1 for error, 0 for success
 */
int MQTTClient::open_session(char* id) {
    /*variable header*/
    char var_header[] = {0x00,0x06,0x4d,0x51,0x49,0x73,0x64,0x70,0x03,0x02,0x00,keepalive/500,0x00,strlen(id)};

    /*fixed header: 2 bytes, big endian*/
    char fixed_header[] = {MQTTCONNECT,12+strlen(id)+2};
    short usernameLen = strlen( userName );
    short passwordLen = strlen( password );
    var_header[9] |= (usernameLen > 0 ? 0x80 : 0x00 );
    var_header[9] |= (passwordLen > 0 ? 0x40 : 0x00 );

    printf("fixed %d, var %d id %d, username %d, password %d\n",sizeof(fixed_header), sizeof(var_header), strlen(id), usernameLen, passwordLen );
    char packet[sizeof(fixed_header) + sizeof(var_header) + strlen(id) + usernameLen + (usernameLen > 0 ? 2 : 0) + passwordLen + (passwordLen > 0 ? 2 : 0) ];

    memset(packet,0,sizeof(packet));
    memcpy(packet,fixed_header,sizeof(fixed_header));
    memcpy(packet+sizeof(fixed_header),var_header,sizeof(var_header));
    memcpy(packet+sizeof(fixed_header)+sizeof(var_header), id, strlen(id));
    if ( usernameLen > 0 ) {
        packet[sizeof(fixed_header)+sizeof(var_header)+strlen(id)] = 0x00;
        packet[sizeof(fixed_header)+sizeof(var_header)+strlen(id)+1] = usernameLen;
        memcpy(packet+sizeof(fixed_header)+sizeof(var_header)+strlen(id)+2, userName, usernameLen);
        packet[1] += (usernameLen + 2);
    }
    if ( passwordLen > 0 ) {
        packet[sizeof(fixed_header)+sizeof(var_header)+strlen(id) + usernameLen + (usernameLen > 0 ? 2 : 0) ] = 0x00;
        packet[sizeof(fixed_header)+sizeof(var_header)+strlen(id) + usernameLen + (usernameLen > 0 ? 2 : 0)+1] = passwordLen;
        memcpy(packet+sizeof(fixed_header)+sizeof(var_header)+strlen(id)+ usernameLen + (usernameLen > 0 ? 2 : 0)+2, password, passwordLen);
        packet[1] += (passwordLen + 2);
    }

/*        printf("Packet length %d\n", sizeof(packet));
        for (int i = 0; i < sizeof(packet); i++) {
            printf("%x, ", packet[i]);
        }
        printf("\r\n");*/
    
    if (send_data(packet, sizeof(packet))) {
        return -1;
    }
    printf("Sent\n");
    return 0;
}

/** Open TCP port, connect to server on given IP address.
 *
 * @returns -1: If connect to server failed. -2: Failed to open session on server.  0: Connection accessed.
 */
int MQTTClient::connect() {
    return connect(clientId);
}    

/** Open TCP port, connect to server on given IP address.
 *
 * @param id The client name shown on MQTT server.
 * @returns -1: If connect to server failed. -2: Failed to open session on server.  0: Connection accessed.
 */
int MQTTClient::connect(char* id) {
    /*Initial TCP socket*/
    if(!pTCPSocket) {
        pTCPSocket = new TCPSocket;
        pTCPSocket->setOnEvent(this, &MQTTClient::onTCPSocketEvent);
    }        

    /*Trying to connect to host*/
    printf("Trying to connect to host..\r\n\r\n");
    TCPSocketErr err = pTCPSocket->connect(host);

    Net::poll();
    if (err) {
        printf("Error connecting to host [%d]\r\n", (int) err);
        return -1;
    }
    printf("Connect to host..\r\n\r\n");

    /*Wait TCP connection with server to be established*/
    int i = 0;
    while (!connected) {
        Net::poll();
        wait(1);
        i++;
        printf("Wait for connections %d..\r\n", i);
        if (i == 35) {//If wait too long, give up.
            return -1;
        }
    }

    /*Send open session message to server*/
    if(open_session(id))    {
        printf("Error open session [%s]\r\n", id);
        return -1;            
    }

    /*Wait server notice of open sesion*/
    i = 0;
    while (!sessionOpened) {
        Net::poll();
        wait(1);
        if (!connected || ++i >40) {
            break;
        }
        printf("Wait for session %d..\r\n", i);
    }
//  if (!connected) {       wakwak_koba
    if (!connected || !sessionOpened)   {
        return -2;
    }
    lastActivity = timer.read_ms();
    return 0;
}

/** Publish a message on a topic.
 *
 * @param pub_topic  The topic name the massage will be publish on.
 * @param msg        The massage to be published.
 * @param retain     Retain flag.
 * @returns  -1: Failed to publish message. 0: Publish sucessed.
 */
int MQTTClient::publish(char* pub_topic, char* msg, bool retain) {
    uint8_t var_header_pub[strlen(pub_topic)+2];
    strcpy((char *)&var_header_pub[2], pub_topic);
    var_header_pub[0] = 0;
    var_header_pub[1] = strlen(pub_topic);

//  uint8_t fixed_header_pub[] = {MQTTPUBLISH,sizeof(var_header_pub)+strlen(msg)};
    int  var_len = sizeof(var_header_pub)+strlen(msg);
    uint8_t fixed_header_pub[1 + (var_len < 128 ? 1 : 2)];
    fixed_header_pub[0] = MQTTPUBLISH | (retain ? 0x01 : 0x00);

    if(var_len < 128)
        fixed_header_pub[1] = var_len;
    else    {
        fixed_header_pub[1] = (var_len % 0x80) | 0x80;
        fixed_header_pub[2] = (var_len / 0x80);
    }
    
    uint8_t packet_pub[sizeof(fixed_header_pub)+sizeof(var_header_pub)+strlen(msg)];
    memset(packet_pub,0,sizeof(packet_pub));
    memcpy(packet_pub,fixed_header_pub,sizeof(fixed_header_pub));
    memcpy(packet_pub+sizeof(fixed_header_pub),var_header_pub,sizeof(var_header_pub));
    memcpy(packet_pub+sizeof(fixed_header_pub)+sizeof(var_header_pub),msg,strlen(msg));

    if (send_data((char*)packet_pub, sizeof(packet_pub))) {
        return -1;
    }
    
    return 0;
}

/** Disconnect from server
 */
void MQTTClient::disconnect() {
    char packet_224[] = {0xe0, 0x00};
    send_data((char*)packet_224, 2);

    pTCPSocket->close();
    delete pTCPSocket;
    pTCPSocket = NULL;
    connected = false;
}

/** Read data from receive packet
 * Determine what needs to be done with packet.
 */
void MQTTClient::read_data() {
    char buffer[1024];
    int len = 0, readLen;

    while ((readLen = pTCPSocket->recv(buffer, 1024)) != 0) {
        len += readLen;
    }

    buffer[len] = '\0';


/*    printf("Read length: %d %d\r\n", len, readLen);

    for (int i = 0; i < len; i++) {
        printf("%2X ", buffer[i]);
    }
    printf("\r\n"); */

    char type = buffer[0]>>4;
    if ( type == 2 ) { // CONNACK
        printf("CONNACK\n");

    } else if (type == 3) { // PUBLISH
        if (callback_server) {
            short index = 1;
            short multiplier = 1;
            short value = 0;
            uint8_t  digit;
            do {
                digit = buffer[index++];
                value += (digit & 127) * multiplier;
                multiplier *= 128;
            } while ((digit & 128) != 0);
            printf( "variable length %d, index %d\n", value, index );

//            uint8_t tl = (buffer[2]<<3)+buffer[3];
            uint8_t tl = (buffer[index]<<3)+buffer[index+1];
            printf("Topic len %d\n",tl);
            char topic[tl+1];
            for (int i=0; i<tl; i++) {
                topic[i] = buffer[index+2+i];
            }
            topic[tl] = 0;
            // ignore msgID - only support QoS 0 subs
            char *payload = buffer+index+2+tl;
            callback_server(topic,(char*)payload);
        }
    } else if (type == 12) { // PINGREG -- Ask for alive
        char packet_208[] = {0xd0, 0x00};
        send_data((char*)packet_208, 2);
        lastActivity = timer.read_ms();
    }
}

/** Check for session opened
 */
void MQTTClient::read_open_session() {
    char buffer[32];
    int len = 0, readLen;

    while ((readLen = pTCPSocket->recv(buffer, 32)) != 0) {
        len += readLen;
    }

    if (len == 4 && buffer[3] == 0) {
        printf("Session opened\r\n");
        sessionOpened = true;
    }
}

/** Subscribe to a topic
 *
 * @param topic The topic name to be subscribed.
 * @returns  -1: Failed to subscribe to topic. 0: Subscribe sucessed.
 */
int MQTTClient::subscribe(char* topic) {

    if (connected) {
        uint8_t var_header_topic[] = {0,10};
        uint8_t fixed_header_topic[] = {MQTTSUBSCRIBE,sizeof(var_header_topic)+strlen(topic)+3};

        //utf topic
        uint8_t utf_topic[strlen(topic)+3];
        strcpy((char *)&utf_topic[2], topic);

        utf_topic[0] = 0;
        utf_topic[1] = strlen(topic);
        utf_topic[sizeof(utf_topic)-1] = 0;

        char packet_topic[sizeof(var_header_topic)+sizeof(fixed_header_topic)+strlen(topic)+3];
        memset(packet_topic,0,sizeof(packet_topic));
        memcpy(packet_topic,fixed_header_topic,sizeof(fixed_header_topic));
        memcpy(packet_topic+sizeof(fixed_header_topic),var_header_topic,sizeof(var_header_topic));
        memcpy(packet_topic+sizeof(fixed_header_topic)+sizeof(var_header_topic),utf_topic,sizeof(utf_topic));

        if (send_data(packet_topic, sizeof(packet_topic))) {
            return -1;
        }
        return 0;
    }
    return -1;
}

/** Send heartbeat/keep-alive message
 *
 */
void MQTTClient::live() {
    if (connected) {
        int t = timer.read_ms();
        if (t - lastActivity > keepalive) {
            //Send 192 0 to broker
            //printf("Send 192\r\n");
            char packet_192[] = {0xc0, 0x00};
            send_data((char*)packet_192, 2);
            lastActivity = t;
        }
    }
}

void MQTTClient::pool()  {
    Net::poll();
}

/** TCP Socket event handling
 *
 * @param e  Event object
 */
void MQTTClient::onTCPSocketEvent(TCPSocketEvent e) {
    switch (e) {
        case TCPSOCKET_ACCEPT:
            printf("New TCPSocketEvent: TCPSOCKET_ACCEPT\r\n");
            break;
        case TCPSOCKET_CONNECTED:
            printf("New TCPSocketEvent: TCPSOCKET_CONNECTED\r\n");
            connected = true;
            break;
        case TCPSOCKET_WRITEABLE:
//            printf("New TCPSocketEvent: TCPSOCKET_WRITEABLE\r\n");
            writtable = true;
            break;
        case TCPSOCKET_READABLE:
            printf("New TCPSocketEvent: TCPSOCKET_READABLE\r\n");
            if (!sessionOpened) {
                read_open_session();
            }
            read_data();
            break;
        case TCPSOCKET_CONTIMEOUT:
            printf("New TCPSocketEvent: TCPSOCKET_CONTIMEOUT\r\n");
            break;
        case TCPSOCKET_CONRST:
            printf("New TCPSocketEvent: TCPSOCKET_CONRST\r\n");
            break;
        case TCPSOCKET_CONABRT:
            printf("New TCPSocketEvent: TCPSOCKET_CONABRT\r\n");
            break;
        case TCPSOCKET_ERROR:
            printf("New TCPSocketEvent: TCPSOCKET_ERROR\r\n");
            break;
        case TCPSOCKET_DISCONNECTED:
            printf("New TCPSocketEvent: TCPSOCKET_DISCONNECTED\r\n");
            pTCPSocket->close();
            connected = false;
            break;
    }
}