Move agnostic MQTT functions to library (#95)

This is the first change in encapsulating the MQTT functionality independently from the hardware.

Author: gguuss

examples/Esp32-lwmqtt/Esp32-lwmqtt.ino

@@ -30,9 +30,11 @@ void loop() {
     connect();
   }
 
+  // TODO: replace with your code
   // publish a message roughly every second.
-  if (millis() - lastMillis > 1000) {
+  if (millis() - lastMillis > 60000) {
     lastMillis = millis();
+    //publishTelemetry(mqttClient, "/sensors", getDefaultSensor());
     publishTelemetry(getDefaultSensor());
   }
-}
+}

examples/Esp32-lwmqtt/esp32-mqtt.h

@@ -16,21 +16,29 @@
 #ifndef __ESP32_MQTT_H__
 #define __ESP32_MQTT_H__
 #include <String.h>
+#include <Client.h>
 #include <WiFi.h>
 #include <WiFiClientSecure.h>
 
-#include <rBase64.h>
 #include <MQTT.h>
 
 #include <CloudIoTCore.h>
+#include <CloudIoTCoreMqtt.h>
 #include "ciotc_config.h" // Update this file with your configuration
 
-// Initialize the Genuino WiFi SSL client library / RTC
-WiFiClientSecure *netClient;
-MQTTClient *mqttClient;
+// !!REPLACEME!!
+// The MQTT callback function for commands and configuration updates
+// Place your message handler code here.
+void messageReceived(String &topic, String &payload) {
+  Serial.println("incoming: " + topic + " - " + payload);
+}
+///////////////////////////////
 
-// Clout IoT configuration that you don't need to change
+// Initialize WiFi and MQTT for this board
+Client *netClient;
 CloudIoTCoreDevice *device;
+CloudIoTCoreMqtt *mqtt;
+MQTTClient *mqttClient;
 unsigned long iss = 0;
 String jwt;
 
@@ -70,159 +78,24 @@ void setupWifi() {
   }
 }
 
-
-///////////////////////////////
-// MQTT common functions
-///////////////////////////////
-void messageReceived(String &topic, String &payload) {
-  Serial.println("incoming: " + topic + " - " + payload);
-}
-
-void startMQTT() {
-  mqttClient->begin("mqtt.googleapis.com", 8883, *netClient);
-  mqttClient->onMessage(messageReceived);
-}
-
-void publishTelemetry(String data) {
-  mqttClient->publish(device->getEventsTopic(), data);
-}
-
-// Helper that just sends default sensor
-void publishState(String data) {
-  mqttClient->publish(device->getStateTopic(), data);
-}
-
-// FIXME: Move to config?
-int __backoff__ = 1000; // current backoff, milliseconds
-int __factor__ = 2.5f;
-int __minbackoff__ = 1000; // minimum backoff, ms
-int __max_backoff__ = 60000; // maximum backoff, ms
-int __jitter__ = 500; // max random jitter, ms
-void mqttConnect() {
-  Serial.print("\nconnecting...");
-  bool keepgoing = true;
-  while (keepgoing) {
-    mqttClient->connect(device->getClientId().c_str(), "unused", getJwt().c_str(), false);
-
-    if (mqttClient->lastError() != LWMQTT_SUCCESS){
-      Serial.println(mqttClient->lastError());
-      switch(mqttClient->lastError()) {
-        case (LWMQTT_BUFFER_TOO_SHORT):
-          Serial.println("LWMQTT_BUFFER_TOO_SHORT");
-          break;
-        case (LWMQTT_VARNUM_OVERFLOW):
-          Serial.println("LWMQTT_VARNUM_OVERFLOW");
-          break;
-        case (LWMQTT_NETWORK_FAILED_CONNECT):
-          Serial.println("LWMQTT_NETWORK_FAILED_CONNECT");
-          break;
-        case (LWMQTT_NETWORK_TIMEOUT):
-          Serial.println("LWMQTT_NETWORK_TIMEOUT");
-          break;
-        case (LWMQTT_NETWORK_FAILED_READ):
-          Serial.println("LWMQTT_NETWORK_FAILED_READ");
-          break;
-        case (LWMQTT_NETWORK_FAILED_WRITE):
-          Serial.println("LWMQTT_NETWORK_FAILED_WRITE");
-          break;
-        case (LWMQTT_REMAINING_LENGTH_OVERFLOW):
-          Serial.println("LWMQTT_REMAINING_LENGTH_OVERFLOW");
-          break;
-        case (LWMQTT_REMAINING_LENGTH_MISMATCH):
-          Serial.println("LWMQTT_REMAINING_LENGTH_MISMATCH");
-          break;
-        case (LWMQTT_MISSING_OR_WRONG_PACKET):
-          Serial.println("LWMQTT_MISSING_OR_WRONG_PACKET");
-          break;
-        case (LWMQTT_CONNECTION_DENIED):
-          Serial.println("LWMQTT_CONNECTION_DENIED");
-          break;
-        case (LWMQTT_FAILED_SUBSCRIPTION):
-          Serial.println("LWMQTT_FAILED_SUBSCRIPTION");
-          break;
-        case (LWMQTT_SUBACK_ARRAY_OVERFLOW):
-          Serial.println("LWMQTT_SUBACK_ARRAY_OVERFLOW");
-          break;
-        case (LWMQTT_PONG_TIMEOUT):
-          Serial.println("LWMQTT_PONG_TIMEOUT");
-          break;
-        default:
-          Serial.println("This error code should never be reached.");
-          break;
-      }
-
-      Serial.println(mqttClient->returnCode());
-      switch(mqttClient->returnCode()) {
-        case (LWMQTT_CONNECTION_ACCEPTED):
-          Serial.println("OK");
-          break;
-        case (LWMQTT_UNACCEPTABLE_PROTOCOL):
-          Serial.println("LWMQTT_UNACCEPTABLE_PROTOCOLL");
-          break;
-        case (LWMQTT_IDENTIFIER_REJECTED):
-          Serial.println("LWMQTT_IDENTIFIER_REJECTED");
-          break;
-        case (LWMQTT_SERVER_UNAVAILABLE):
-          Serial.println("LWMQTT_SERVER_UNAVAILABLE");
-          break;
-        case (LWMQTT_BAD_USERNAME_OR_PASSWORD):
-          Serial.println("LWMQTT_BAD_USERNAME_OR_PASSWORD");
-          iss = 0; // Force JWT regeneration
-          break;
-        case (LWMQTT_NOT_AUTHORIZED):
-          Serial.println("LWMQTT_NOT_AUTHORIZED");
-          iss = 0; // Force JWT regeneration
-          break;
-        case (LWMQTT_UNKNOWN_RETURN_CODE):
-          Serial.println("LWMQTT_UNKNOWN_RETURN_CODE");
-          break;
-        default:
-          Serial.println("This return code should never be reached.");
-          break;
-      }
-      // See https://cloud.google.com/iot/docs/how-tos/exponential-backoff
-      if (__backoff__ < __minbackoff__) {
-        __backoff__ = __minbackoff__;
-      }
-      __backoff__ = (__backoff__ * __factor__) + random(__jitter__);
-      if (__backoff__ > __max_backoff__) {
-        __backoff__ = __max_backoff__;
-      }
-
-      // Clean up the client
-      mqttClient->disconnect();
-      Serial.println("Delaying " + String(__backoff__) + "ms");
-      delay(__backoff__);
-      keepgoing = true;
-    } else {
-      // We're now connected
-      Serial.println("\nconnected!");
-      keepgoing = false;
-      __backoff__ = __minbackoff__;
-    }
-  }
-
-  mqttClient->subscribe(device->getConfigTopic(), 1); // Set QoS to 1 (ack) for configuration messages
-  mqttClient->subscribe(device->getCommandsTopic(), 0); // QoS 0 (no ack) for commands
-  if (ex_num_topics > 0) { // Subscribe to the extra topics
-    for (int i=0; i < ex_num_topics; i++) {
-        mqttClient->subscribe(ex_topics[i], 0); // QoS 0 (no ack) for commands
-    }
+void connectWifi() {
+  Serial.print("checking wifi...");
+  while (WiFi.status() != WL_CONNECTED) {
+    Serial.print(".");
+    delay(1000);
   }
-
-  publishState("connected");
 }
 
 ///////////////////////////////
 // Orchestrates various methods from preceeding code.
 ///////////////////////////////
+void publishTelemetry(String data) {
+  mqtt->publishTelemetry(data);
+}
+
 void connect() {
-  Serial.print("checking wifi...");
-  while (WiFi.status() != WL_CONNECTED) {
-    Serial.print(".");
-    delay(1000);
-  }
-  mqttConnect();
+  connectWifi();
+  mqtt->mqttConnect();
 }
 
 void setupCloudIoT() {
@@ -234,6 +107,7 @@ void setupCloudIoT() {
   netClient = new WiFiClientSecure();
   mqttClient = new MQTTClient(512);
   mqttClient->setOptions(180, true, 1000); // keepAlive, cleanSession, timeout
-  startMQTT();
+  mqtt = new CloudIoTCoreMqtt(mqttClient, netClient, device);
+  mqtt->startMQTT();
 }
 #endif //__ESP32_MQTT_H__

examples/Esp8266-lwmqtt/Esp8266-lwmqtt.ino

@@ -1,43 +1,44 @@
-/******************************************************************************
- * Copyright 2018 Google
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *****************************************************************************/
-#include "esp8266_mqtt.h"
-
-#ifndef LED_BUILTIN
-  #define LED_BUILTIN 13
-#endif
-
-void setup() {
-  // put your setup code here, to run once:
-  Serial.begin(115200);
-  setupCloudIoT(); // Creates globals for MQTT
-  pinMode(LED_BUILTIN, OUTPUT);
-  startMQTT();
-}
-
-unsigned long lastMillis = 0;
-void loop() {
-  mqttClient->loop();
-  delay(10);  // <- fixes some issues with WiFi stability
-
-  if (!mqttClient->connected()) {
-    connect();
-  }
-
-  // publish a message roughly every second.
-  if (millis() - lastMillis > 1000) {
-    lastMillis = millis();
-    publishTelemetry(getDefaultSensor());
-  }
+/******************************************************************************
+ * Copyright 2018 Google
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *****************************************************************************/
+#include <CloudIoTCore.h>
+
+#include "esp8266_mqtt.h"
+
+#ifndef LED_BUILTIN
+  #define LED_BUILTIN 13
+#endif
+
+void setup() {
+  // put your setup code here, to run once:
+  Serial.begin(115200);
+  setupCloudIoT(); // Creates globals for MQTT
+  pinMode(LED_BUILTIN, OUTPUT);
+}
+
+unsigned long lastMillis = 0;
+void loop() {
+  mqttClient->loop();
+  delay(10);  // <- fixes some issues with WiFi stability
+
+  if (!mqttClient->connected()) {
+    connect();
+  }
+
+  // TODO: Replace with your code here
+  if (millis() - lastMillis > 60000) {
+    lastMillis = millis();
+    publishTelemetry(getDefaultSensor());
+  }
 }