NetMeterbusMaster/MqttClient.cpp

/*
 * MqttClient.cpp
 *
 *  Created on: 08.05.2015
 *      Author: wn
 */

#include "MqttClient.h"

#include <PubSubClient.h>
#include <PString.h>
#include <Streaming.h>
#include <Metro.h>


byte MQTT_BROKER[] = { 172, 16, 2, 16 };
const uint16_t MQTT_PORT = 1883;


void callback(char* topic, byte* payload, unsigned int length) {
}


Metro secondTick = Metro(1000);


MqttClient::MqttClient(RequestSender *meterBusMaster) :
                m_client(), m_meterBusMaster(meterBusMaster), m_mqttClient(MQTT_BROKER, MQTT_PORT, callback, m_client),
                m_disconnectState(3), m_disconnectTime(millis()), m_uptime(0), m_deviceIdx(0)
{
  for (uint8_t i = 0; i < NUM_OF_DEVICES; i++) {
    m_mbusDevTuple[i] = { 0, 0, 0, 0 };
  }

  m_mbusDevTuple[0] = { 1, 0x53, 10, 0 }; // light meter
  m_mbusDevTuple[1] = { 2, 32,   10, 0 }; // electrity
}

void MqttClient::sendResponse(uint8_t *responseBuffer, uint16_t responseBufferLength, uint8_t token) {
  char strbuf[1024];
  memset(strbuf, sizeof(strbuf), 0);
  PString buf = PString(strbuf, sizeof(strbuf));


  buf << "{ \"metadata\": { \"device\": \"MeterbusHub\", " <<
      "\"token\": " << token << ", " <<
      "}, " <<
      "\"data\": {" <<
      "\"uptime\": " << m_uptime << ", " <<
      "\"telegram\": \"";

  uint16_t i = 0;
  while (true) {
    if (responseBuffer[i] <= 0x0f) {
      buf.print("0");
    }
    buf.print(responseBuffer[i], HEX);
    buf.print(" ");
    i++;
    if (i == responseBufferLength) {
      break;
    }
  }

  buf << "\"}}";

  if (m_disconnectState == 0) {
    //Serial << "publishing " << strbuf << endl;
    //Serial << "length: " << buf.length() << endl;
    m_mqttClient.publish("MeterbusHub/Measurement", strbuf);
  } else {
    Serial << "no MQTT connection, message lost: " << endl <<
        strbuf << endl;
  }
}

void MqttClient::sendError(uint8_t code, uint8_t token) {
  String msg = String("{ \"metadata\": { \"device\": \"MeterbusHub\", \"error\": ")
      + code + String(", \"token\": ") + token + String(" }, \"data\": { \"uptime\": ") + m_uptime + String("}}");
  if (m_disconnectState == 0) {
    //Serial << "publishing " << msg << endl;
    //Serial << "length: " << msg.length() << endl;
    m_mqttClient.publish("MeterbusHub/Measurement", (char*)msg.c_str());
  } else {
    Serial << "no MQTT connection, message lost: " << msg << endl;
  }
}

void MqttClient::begin() {
}

void MqttClient::exec() {
  //Serial << "*** a" << endl;
  if ((m_disconnectState == 0) && (! m_mqttClient.loop())) {
    m_disconnectState = 1;
  }

  //Serial << "*** b, " << m_disconnectState << endl;
  switch (m_disconnectState) {
  case 0:
    // Serial.println("discState 0");
    // everything fine
    break;
  case 1:
    Serial.println("MQTT Connection lost");
    m_mqttClient.disconnect();
    m_disconnectTime = millis();
    m_disconnectState = 2;
    break;
  case 2:
    if (m_disconnectTime + 2000 < millis()) {
      m_disconnectState = 3;
    }
    break;
  case 3:
    Serial << "Trying to re-connect" << endl;
    if (m_mqttClient.connect("MeterbusHub")) {
      Serial << "MQTT connected" << endl;
      m_disconnectTime = millis();
      m_disconnectState = 0;
    } else {
      Serial << "no success" << endl;
      m_disconnectState = 1;
    }
    break;
  default:
    m_disconnectState = 0;
    break;
  }

  //Serial << "*** c" << endl;
  if (secondTick.check() == 1) {
    m_uptime++;
    //Serial << "Tick " << m_uptime << endl;

    String msg = String("{ \"metadata\": { \"device\": \"MeterbusHub\" }, \"data\": { \"uptime\": ") + m_uptime + String("}}");
    if (m_disconnectState == 0) {
      m_mqttClient.publish("MeterbusHub/Heartbeat", (char*)msg.c_str());
    } else {
      Serial << "no MQTT connection, message lost: " << msg << endl;
    }

    for (uint8_t i = 0; i < NUM_OF_DEVICES; i++) {
      if ((m_mbusDevTuple[i].address != 0) && (m_mbusDevTuple[i].timer != 0)) {
        m_mbusDevTuple[i].timer -= 1;
//        if (m_mbusDevTuple[i].timer == 0) {
//          Serial << "Device " << m_mbusDevTuple[i].token << " ready for request" << endl;
//        } else {
//          Serial << "Device " << m_mbusDevTuple[i].token << " not ready for request, timer: " << m_mbusDevTuple[i].timer << endl;
//        }
      }
    }

    //Serial << "while in" << endl;
    while (true) {
      if ((m_mbusDevTuple[m_deviceIdx].address != 0) && (m_mbusDevTuple[m_deviceIdx].timer == 0)) {
        Serial << "Issue request for device " << m_mbusDevTuple[m_deviceIdx].token << endl;
        uint8_t *sendBuffer = m_meterBusMaster->getSendBuffer();
        if (sendBuffer != 0) {
          Serial << "send buffer ready" << endl;
          m_mbusDevTuple[m_deviceIdx].timer = m_mbusDevTuple[m_deviceIdx].queryPeriod;

          sendBuffer[0] = 0x10;
          sendBuffer[1] = 0x5b;
          sendBuffer[2] = m_mbusDevTuple[m_deviceIdx].address;
          sendBuffer[3] = (uint8_t)(sendBuffer[1] + sendBuffer[2]);
          sendBuffer[4] = 0x16;
          m_meterBusMaster->sendBufferReady(5, m_mbusDevTuple[m_deviceIdx].token, this);
        } else {
          Serial << "no send buffer ready" << endl;
        }
        break;
      } else {
        //Serial << "Trying " << m_deviceIdx << ", " << m_mbusDevTuple[m_deviceIdx].token << endl;
        m_deviceIdx++;
        if (m_deviceIdx >= NUM_OF_DEVICES) {
          m_deviceIdx = 0;
        }
        break;
      }
    }
    //Serial << "while out" << endl;
  }

  //Serial << "*** d" << endl;

}