File content as of revision 13:65e87bd958bd:

/*******************************************************************************
* Copyright (c) 2014 IBM Corporation and other Contributors.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* which accompanies this distribution, and is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* Contributors: Sam Danbury
* IBM - Initial Contribution
*******************************************************************************/

#include "ExampleClient.h"

int i =0;
ExampleClient::ExampleClient(string mac) {
    quickstartMode = true;
    connected = false;
    macAddress = mac;
    
    //Generate topic string for publish
    memcpy(topic, "iot-2/evt/status/fmt/json", 25);
    topic[25] = '\0';
    
    //Generate topic string for subscribe
    memcpy(subscribeTopic, "iot-2/cmd/blink/fmt/json",24);
    subscribeTopic[24] = '\0';
    
    loadConfig();
    
    tryMqttConnect();
}

void ExampleClient::loadConfig() {
    
    ConfigFile cfg;
    
    char value[30];
    char value1[30];
    char value2[30];
    char value3[30];
    char value4[30];
    if (cfg.read("/local/device.cfg")) {
        quickstartMode = false;
        
        if (cfg.getValue("org", value, sizeof(value))) {
            stringstream ss(value);
            ss >> org;
        } else {
            lcd.printf("No org defined in config\n");
        }
        
        if (cfg.getValue("type", value1, sizeof(value1))) {
            stringstream ss(value1);
            ss >> type;
        } else {
            lcd.printf("No type defined in config\n");
        }
        
        if (cfg.getValue("id", value2, sizeof(value2))) {
            stringstream ss(value2);
            ss >> id;
        } else {
            lcd.printf("No id defined in config\n");
        }
        
        if (cfg.getValue("auth-method", value3, sizeof(value3))) {
            stringstream ss(value3);
            ss >> authMethod;
        } else {
            lcd.printf("No auth method defined in config\n");
        }
        
        if (cfg.getValue("auth-token", value4, sizeof(value4))) {
            stringstream ss(value4);
            ss >> token;
        } else {
            lcd.printf("No token defined in config\n");
        }
        
    } else {
        org = "quickstart";
        type = "iotsample-mbed-lpc1768";
        id = macAddress;
    }
    wait(5.0);
    
}

int ExampleClient::reconnectDelay(int i) {
    if (i < 10) {
        return 3; //First 10 attempts try within 3 seconds
    } else if (i < 20) {
        return 60; //Next 10 attempts retry after every 1 minute
    } else {
        return 600; //After 20 attempts, retry every 10 minutes
    }
}

void ExampleClient::tryMqttConnect() {
    int retryAttempt = 0;
    
    //Reinstantiate TCP socket connection object
    mysock = TCPSocketConnection();
    
    while (connected == false) {
        lcd.cls();
        lcd.locate(0,0);
        lcd.printf("Trying to connect...");
        
        //Based on number of connection attempts, determine timeout
        int connDelayTimeout = reconnectDelay(++retryAttempt);
        
        //Attempt to reconnect
        connect();
        
        //If connection was not established, continue retry
        if (connected == false) {
            wait(connDelayTimeout);
        } else {
            break;
        }
    }
}

void ExampleClient::connect() {
    MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
    int  rc = 0;
    int  len = 0;
    char buf[200];
    int  buflen = sizeof(buf);
    
    //Connect to TCP socket
    mysock.connect(IBM_IOT_BROKER, IBM_IOT_PORT);
    
    //Construct client ID
    string str = string("d:") + org + ":" + type + ":" + id;
    char clientId[str.size()+1];
    memcpy(clientId, str.c_str(), str.size() + 1);
    
    //Set MQTT connect options
    data.clientID.cstring = clientId;
    data.keepAliveInterval = 20;
    data.cleansession = 1;
    data.MQTTVersion = 3;
    if (!quickstartMode) {
        if ( ! authMethod.compare("token") ) {
            data.username.cstring = clientId;
            char* authToken= new char[token.size()];
            memcpy(authToken, token.c_str(), token.size()+1);
            data.password.cstring =  authToken;
        }
        else {
            lcd.printf("Exiting\n");
            exit(-1);
        }
    }
    
    //Attempt MQTT connect
    len = MQTTSerialize_connect(buf, buflen, &data);
    rc = 0;
    while (rc < len) {
        int rc1 = mysock.send(buf, len);
        if (rc1 == -1) {
            connected = false;
            break;
        } else {
            rc += rc1;
        }
    }
    if (rc == len) {
        connected = true;
    }
    wait(0.2);
}

void ExampleClient::publish(string thePayload) {
    int  rc = 0;
    int  len = 0;
    char buf[250];
    int  buflen = sizeof(buf);

    MQTTString topicString = MQTTString_initializer;
    
    topicString.cstring = topic;
    
    //Convert payload from string to char array
    char* payload = new char [thePayload.length()+1];
    std::strcpy (payload, thePayload.c_str());
    int payloadlen = strlen(payload);
    //Attempt MQTT publish
    len = MQTTSerialize_publish(buf, buflen, 0, 0, 0, 0, topicString, payload, payloadlen);
    rc = 0;
    while (rc < len) {
        int rc1 = mysock.send(buf, len);
        wait(3.0);
        if (rc1 == -1) {
            //If return code from MQTT publish is -1, attempt reconnect
            connected = false;
            tryMqttConnect();
            break;
        } else {
            rc += rc1;
        }
    }
    wait(0.2);
    
    if (payload) {
        delete payload;
    }
}

int ExampleClient::subscribe() {
    int  rc = 0;
    int  len = 0;
    char buf[250];
    int  buflen = sizeof(buf);

    MQTTString topicString = MQTTString_initializer;

    topicString.cstring = subscribeTopic;

    //Attempt MQTT subscribe
    len = MQTTSerialize_subscribe(buf, buflen, 0, 1, 1, &topicString, 0);
    rc = 0;
    while (rc < len) {
        int rc1 = mysock.send(buf, len);
        if (rc1 == -1) {
            break;
        } 
        else {
            rc += rc1;
        }
    }
    wait(0.2);
    return rc;
}