SinricPro with Mqtt PubSub

[peastuti]

Hello There,

I'm working on the integration of Sinric Pro in one of my IoT Device, this particular one is quite tricky. Using the PubSub library, it establishes an mqtt connection to a AWS IoT core mqtt queue. In the meanwhile, it was able to work with standard sinric and, consequently, the integration with Google Home.

Maybe it is uncorrelated, but it is worth trying and ask. There could be some kind of interference on these two libraries?

It seems that the loop() function is unable to handle both the mqtt client and the SinricPro.handle()

This is the example from which I took inspiration with my mqtt part https://github.com/debsahu/ESP-MQTT-AWS-IoT-Core/blob/master/Arduino/PubSubClient/PubSubClient.ino

Also, this is my loop

void loop() {
  SinricPro.handle();
  handleTemperaturesensor();

  if(!client.connected()){
    checkWiFiThenMQTT();
  }
  else {
    client.loop();
    if(millis() >= timeNowMyPeriod + myPeriod){
      timeNowMyPeriod += myPeriod;
      float finalTemp = getTemperature();
      sendData(finalTemp);
    }
  }

  ArduinoOTA.handle();
  Debug.handle();

}

In this particular case, none of the two service is able to connect, while If I stop looping on the mqtt part (commenting the IF), the SinricPro loop works correctly.

Any suggestion? Maybe I'm missing something.

[sivar2311]

I did a test using PubSubClient on Adafruit.io and SinricPro. There are no interferences. Everything works flawlessly.

void handlePubSub() {
  static unsigned long lastTry;
  unsigned long currentMillis = millis();
  if (!client.connected() && currentMillis - lastTry > 5000) {
    lastTry = currentMillis;
    Serial.printf("Attempting MQTT connection...");

    if (client.connect("ESP8266-Test", MQTT_USERNAME, MQTT_KEY)) {
      Serial.printf("connected\r\n");
      client.subscribe(MQTT_FEED);
    } else {
      Serial.printf("failed, rc=%d try again in 5 seconds\r\n", client.state());
    }
  }
  client.loop();
}

void loop() {
  handlePubSub();
  SinricPro.handle();
}

[peastuti]

Thank you for your prompt response. But my issue is still present. It is like something is conflicting. If I move in loop() handlePubSub() before SinricPro.handle() mqtt works but not sinric, and viceversa.

Here's my code, 4 eyes are better than 2, I guess.

#include <Arduino.h>
#include <ESP8266WiFi.h>
#include <WiFiClientSecure.h>

#include <OneWire.h>
#include <DallasTemperature.h>

#include <PubSubClient.h>
#include <time.h>
#include <ArduinoOTA.h>
#include "RemoteDebug.h" 

#include "SinricPro.h"
#include "SinricProTemperaturesensor.h"

#include "secrets.h"
#include "aquarium_config.h"

// SinricPro Variables
#define APP_KEY           ""      // Should look like "de0bxxxx-1x3x-4x3x-ax2x-5dabxxxxxxxx"
#define APP_SECRET        ""   // Should look like "5f36xxxx-x3x7-4x3x-xexe-e86724a9xxxx-4c4axxxx-3x3x-x5xe-x9x3-333d65xxxxxx"
#define TEMP_SENSOR_ID    ""    // Should look like "5dc1564130xxxxxxxxxxxxxx"
#define BAUD_RATE         115200                // Change baudrate to your need
#define EVENT_WAIT_TIME   60000               // send event every 60 seconds
bool deviceIsOn;                              // Temeprature sensor on/off state
float temperature;                            // actual temperature

// Timers Variables
unsigned long lastEvent = (-EVENT_WAIT_TIME); // last time event has been sent
unsigned long lastMillis = 0;
time_t now;
time_t nowish = 1510592825;
unsigned long previousMillis = 0;
const long interval = 5000;
unsigned long myTime;
unsigned long startTime;
uint64_t heartbeatTimestamp = 0;
uint64_t pollTimestamp = 0;

// TemperatureSensors Variable
#define ONE_WIRE_BUS 4
#define emptyString String()
OneWire oneWire(ONE_WIRE_BUS);
DallasTemperature sensors(&oneWire);
int numberOfDevices;
DeviceAddress tempDeviceAddress; 

// MQTT Variables
bool isConnected = false;
WiFiClientSecure net;
RemoteDebug Debug;
const int MQTT_PORT = 8883;
const char MQTT_PUB_TOPIC_DATA[] = "" THINGNAME "/data";
const char MQTT_PUB_TOPIC_HB[] = "" THINGNAME "/heartbeat";
const char CLIENT_ID[] = THINGNAME "-aquarium";
uint8_t DST = 1;
BearSSL::X509List cert(cacert);
BearSSL::X509List client_crt(client_cert);
BearSSL::PrivateKey key(privkey);
PubSubClient client(net);

// Functions Declarations
void sendHeartbeat();
void sendData(float finalTemp);

float getTemperature(){
  sensors.requestTemperatures(); 
  float average = 0;
  for(int i=0;i<numberOfDevices; i++){
    if(sensors.getAddress(tempDeviceAddress, i)){
      float tempC = sensors.getTempC(tempDeviceAddress);
      average += tempC;
    }
  }
  float finalTemp = average / numberOfDevices;
  return finalTemp;
}

void flashInternalLed(int c, int mydelay){
  for(int i=0; i<c; i++){
    digitalWrite(LED_BUILTIN, LOW);
    delay(mydelay);
    digitalWrite(LED_BUILTIN, HIGH);
    delay(mydelay);
  }
}

void NTPConnect(void) {
  Serial.print("Setting time using SNTP"); Debug.print("Setting time using SNTP");
  configTime(TIME_ZONE * 3600, DST * 3600, "pool.ntp.org", "time.nist.gov");
  now = time(nullptr);
  while (now < nowish) {
    flashInternalLed(1, 250);
    Serial.print("."); Debug.print(".");
    now = time(nullptr);
  }
  Serial.println("done!"); Debug.println("done!");
  struct tm timeinfo;
  gmtime_r(&now, &timeinfo);
  Serial.print("Current time: "); Debug.print("Current time: ");
  Serial.print(asctime(&timeinfo)); Debug.print(asctime(&timeinfo));
}

void messageReceived(char *topic, byte *payload, unsigned int length) {
  flashInternalLed(3, 20);

  DynamicJsonDocument doc(length + 100);
  deserializeJson(doc, payload);

  Serial.println("Received Message"); Debug.println("Received Message");
  serializeJson(doc, Serial); serializeJson(doc, Debug);
  Serial.println(); Debug.println();

  String command = doc["command"];

  if (command == "heartbeat"){
    sendHeartbeat();
  }
  if (command == "force-send-data"){
    sendData(getTemperature());
  }
  else {
    Serial.println("No action taken"); Debug.println("No action taken");
  }
  Serial.println(); Debug.println();
}

void pubSubErr(int8_t MQTTErr) {
  if (MQTTErr == MQTT_CONNECTION_TIMEOUT) {
    Serial.print("Connection tiemout"); Debug.print("Connection tiemout"); }
  else if (MQTTErr == MQTT_CONNECTION_LOST) {
    Serial.print("Connection lost"); Debug.print("Connection lost"); }
  else if (MQTTErr == MQTT_CONNECT_FAILED) {
    Serial.print("Connect failed"); Debug.print("Connect failed"); }
  else if (MQTTErr == MQTT_DISCONNECTED) {
    Serial.print("Disconnected"); Debug.print("Disconnected"); }
  else if (MQTTErr == MQTT_CONNECTED) {
    Serial.print("Connected"); Debug.print("Connected"); }
  else if (MQTTErr == MQTT_CONNECT_BAD_PROTOCOL) {
    Serial.print("Connect bad protocol"); Debug.print("Connect bad protocol"); }
  else if (MQTTErr == MQTT_CONNECT_BAD_CLIENT_ID) {
    Serial.print("Connect bad Client-ID"); Debug.print("Connect bad Client-ID"); }
  else if (MQTTErr == MQTT_CONNECT_UNAVAILABLE) {
    Serial.print("Connect unavailable"); Debug.print("Connect unavailable"); }
  else if (MQTTErr == MQTT_CONNECT_BAD_CREDENTIALS) {
    Serial.print("Connect bad credentials"); Debug.print("Connect bad credentials"); }
  else if (MQTTErr == MQTT_CONNECT_UNAUTHORIZED) {
    Serial.print("Connect unauthorized"); Debug.print("Connect unauthorized"); }
}

void connectToMqtt(bool nonBlocking = false) {
  Serial.print("MQTT connecting "); Debug.print("MQTT connecting ");
  while (!client.connected()) {
    flashInternalLed(1, 50);
    if (client.connect(CLIENT_ID)){
      Serial.println("connected!"); Debug.println("connected!");
      flashInternalLed(5, 50);
      if (!client.subscribe(MQTT_PUB_TOPIC_DATA))
        pubSubErr(client.state());
      if (!client.subscribe(MQTT_PUB_TOPIC_HB))
        pubSubErr(client.state());
    }
    else
    {
      Serial.print("failed, reason -> "); Debug.print("failed, reason -> ");
      pubSubErr(client.state());
      if (!nonBlocking) {
        Serial.println(" < try again in 5 seconds"); Debug.println(" < try again in 5 seconds");
        flashInternalLed(5, 500);
      }
      else {
        Serial.println(" <"); Debug.println(" <");
      }
    }
    if (nonBlocking)
      break;
  }
}

void connectToWiFi(String init_str){
  if (init_str != emptyString)
    Serial.print(init_str); Debug.print(init_str);
  while (WiFi.status() != WL_CONNECTED) {
    Serial.print("."); Debug.print(".");
    flashInternalLed(5, 100);
  }
  if (init_str != emptyString)
    Serial.println("ok!"); Debug.println("ok!");
}

void checkWiFiThenMQTT(void) {
  connectToWiFi("Checking WiFi");
  connectToMqtt();
}

void checkWiFiThenMQTTNonBlocking(void) {
  connectToWiFi(emptyString);
  if (millis() - previousMillis >= interval && !client.connected()) {
    previousMillis = millis();
    connectToMqtt(true);
  }
}

void checkWiFiThenReboot(void) {
  connectToWiFi("Checking WiFi");
  Serial.print("Rebooting"); Debug.print("Rebooting");
  ESP.restart();
}

void sendHeartbeat(void){
  DynamicJsonDocument jsonBuffer(JSON_OBJECT_SIZE(3) + 100);
  JsonObject root = jsonBuffer.to<JsonObject>();
  root["deviceId"]      = THINGNAME;
  root["response"]      = "heartbeat-alive";
  root["thing"]         = "aquarium";

  Serial.printf("Sending  [%s]: ", MQTT_PUB_TOPIC_HB); Debug.printf("Sending  [%s]: ", MQTT_PUB_TOPIC_HB);
  serializeJson(root, Serial); serializeJson(root, Debug);
  Serial.println(); Debug.println();
  char shadow[measureJson(root) + 1];
  serializeJson(root, shadow, sizeof(shadow));

  if (!client.publish(MQTT_PUB_TOPIC_HB, shadow, false))
    pubSubErr(client.state());
}

void sendData(float finalTemp) {
  DynamicJsonDocument jsonBuffer(JSON_OBJECT_SIZE(3) + 100);
  JsonObject root = jsonBuffer.to<JsonObject>();

  time_t now; time(&now);
  root["deviceId"]      = THINGNAME;
  root["timestamp"]     = now;
  root["temperature"] = finalTemp;

  Serial.printf("Sending  [%s]: ", MQTT_PUB_TOPIC_DATA); Debug.printf("Sending  [%s]: ", MQTT_PUB_TOPIC_DATA);
  serializeJson(root, Serial); serializeJson(root, Debug);
  Serial.println(); Debug.println();
  char shadow[measureJson(root) + 1];
  serializeJson(root, shadow, sizeof(shadow));

  if (!client.publish(MQTT_PUB_TOPIC_DATA, shadow, false))
    pubSubErr(client.state());
}

void arduinoOtaSetup(){
  ArduinoOTA.onStart([]() {
  String type;
  if (ArduinoOTA.getCommand() == U_FLASH) {
    type = "sketch";
  } else { // U_FS
    type = "filesystem";
  }
  // NOTE: if updating FS this would be the place to unmount FS using FS.end()
  Serial.println("Start updating " + type); Debug.println("Start updating " + type);
  });
  ArduinoOTA.onEnd([]() {
    Serial.println("\nEnd"); Debug.println("\nEnd"); 
  });
  ArduinoOTA.onProgress([](unsigned int progress, unsigned int total) {
    Serial.printf("Progress: %u%%\r", (progress / (total / 100))); Debug.printf("Progress: %u%%\r", (progress / (total / 100)));
  });
  ArduinoOTA.onError([](ota_error_t error) {
    Serial.printf("Error[%u]: ", error); Debug.printf("Error[%u]: ", error);
    if (error == OTA_AUTH_ERROR) {
      Serial.println("Auth Failed"); Debug.println("Auth Failed");
    } else if (error == OTA_BEGIN_ERROR) {
      Serial.println("Begin Failed"); Debug.println("Begin Failed");
    } else if (error == OTA_CONNECT_ERROR) {
      Serial.println("Connect Failed"); Debug.println("Connect Failed");
    } else if (error == OTA_RECEIVE_ERROR) {
      Serial.println("Receive Failed"); Debug.println("Receive Failed");
    } else if (error == OTA_END_ERROR) {
      Serial.println("End Failed"); Debug.println("End Failed");
    }
  });
  ArduinoOTA.begin();
}

bool onPowerState(const String &deviceId, bool &state) {
  Serial.printf("Temperaturesensor turned %s (via SinricPro) \r\n", state?"on":"off");
  deviceIsOn = state; // turn on / off temperature sensor
  return true; // request handled properly
}

void handleTemperaturesensor() {
  if (deviceIsOn == false) return; // device is off...do nothing

  unsigned long actualMillis = millis();
  if (actualMillis - lastEvent < EVENT_WAIT_TIME) return; //only check every EVENT_WAIT_TIME milliseconds

  temperature = 22.1;          // get actual temperature in °C

  SinricProTemperaturesensor &mySensor = SinricPro[TEMP_SENSOR_ID];  // get temperaturesensor device
  bool success = mySensor.sendTemperatureEvent(temperature); // send event
  if (success) {  // if event was sent successfuly, print temperature and humidity to serial
    Serial.printf("Temperature: %2.1f Celsius\r\n", temperature);
  } else {  // if sending event failed, print error message
    Serial.printf("Something went wrong...could not send Event to server!\r\n");
  }

  lastEvent = actualMillis;       // save actual time for next compare
}

// setup function for SinricPro
void setupSinricPro() {
  // add device to SinricPro
  Serial.println("Setting up SinricPro");
  SinricProTemperaturesensor &mySensor = SinricPro[TEMP_SENSOR_ID];
  mySensor.onPowerState(onPowerState);

  // setup SinricPro
  SinricPro.onConnected([](){ Serial.printf("Connected to SinricPro\r\n"); }); 
  SinricPro.onDisconnected([](){ Serial.printf("Disconnected from SinricPro\r\n"); });
  SinricPro.begin(APP_KEY, APP_SECRET);
  SinricPro.restoreDeviceStates(true); // get latest known deviceState from server (is device turned on?)
}

void setup(){
  Serial.begin(BAUD_RATE);

  pinMode(LED_BUILTIN, OUTPUT);

  flashInternalLed(5, 500);

  Serial.println(); Debug.println();
  Serial.println(); Debug.println();
  WiFi.hostname(THINGNAME);
  WiFi.config(ip, dns, gateway, subnet);
  WiFi.mode(WIFI_STA);
  WiFi.begin(ssid, pass);

  connectToWiFi(String("Attempting to connect to SSID: ") + String(ssid));
  Serial.println(WiFi.localIP()); Debug.println(WiFi.localIP());

  Debug.begin(WiFi.localIP().toString());
  setupSinricPro();

  NTPConnect();

  net.setTrustAnchors(&cert);
  net.setClientRSACert(&client_crt, &key);

  arduinoOtaSetup();

  client.setServer(MQTT_HOST, MQTT_PORT);
  client.setCallback(messageReceived);

  // Start up the library
  sensors.begin();
  numberOfDevices = sensors.getDeviceCount();
}

void handlePubSub() {
  static unsigned long lastTry;
  unsigned long currentMillis = millis();
  if (!client.connected() && currentMillis - lastTry > 5000) {
    lastTry = currentMillis;
    Serial.printf("Attempting MQTT connection...");

    if (client.connect(CLIENT_ID)) {
      Serial.printf("connected\r\n");
      client.subscribe(MQTT_PUB_TOPIC_DATA);
    } else {
      Serial.printf("failed, rc=%d try again in 5 seconds\r\n", client.state());
    }
  }
  client.loop();
}

unsigned long myPeriod = 120000;
unsigned long timeNowMyPeriod = 0;
void loop() {
  SinricPro.handle();
  handlePubSub();
  handleTemperaturesensor();

  // ArduinoOTA.handle();
  // Debug.handle();

}

in this case, the output is

Attempting to connect to SSID: lol...ok!
192.168.7.103
Setting up SinricPro
Setting time using SNTP.........................done!
Current time: Sat Feb 27 11:03:14 2021
Attempting MQTT connection...failed, rc=-2 try again in 5 seconds
Connected to SinricPro
Temperaturesensor turned on (via SinricPro) 
Temperature: 22.1 Celsius
Attempting MQTT connection...failed, rc=-2 try again in 5 seconds
Attempting MQTT connection...failed, rc=-2 try again in 5 seconds
Attempting MQTT connection...failed, rc=-2 try again in 5 seconds

as said if I swap the two handles, the SinricHandle is not able to connect at all.

Do you see something that I'm missing?

[sivar2311]

Woh, it's a bit complex for a short "overview".

I suggest to try with a much simpler sketch first - like i did. Just one SinricProSwitch that can be turned on off from both sides (SinricPro / MQTT).

If this sketch is running stable, extend it with the next "feature".

[sivar2311]

I made a simple example for you. The result is shown here

I hope this might help you.

[peastuti]

@sivar2311 that's really a lot of effort. I'm really thankful!! But, something is still missing. You see, to connect my mqtt client I had to made some modification to your example. I list them here for the sake of clarity:

• used different port - 8883 instead of 1883
• used WiFiClientSecure wifiClient instead of WiFiClient wifiClient
• attached my certificates to it

  BearSSL::X509List cert(cacert);
  BearSSL::X509List client_crt(client_cert);
  BearSSL::PrivateKey key(privkey);

• added a NTPConnect method to retrieve time and added it in setup, after wifi connection

  time_t now;
  time_t nowish = 1510592825;
  void NTPConnect(void) {
  Serial.print("Setting time using SNTP"); 
  configTime(TIME_ZONE * 3600, 1 * 3600, "pool.ntp.org", "time.nist.gov");
  now = time(nullptr);
  while (now < nowish) {
  delay(500);
  Serial.print(".");
  now = time(nullptr);
  }
  Serial.println("done!");
  struct tm timeinfo;
  gmtime_r(&now, &timeinfo);
  Serial.print("Current time: "); 
  Serial.print(asctime(&timeinfo));
  }

• then, i changed client.connect("ESP8266-Test", MQTT_USERNAME, MQTT_KEY) with client.connect("ESP8266-Test") since I use the certificates to connect to the MQTT server.

Here's the integral version (which is quite the same as yours but with the modifications listed before)

This way, the mqtt connection start, but no Sinric, as always if I comment out the setup and loops of one of them in a mutually exclusive way, it works 😩😩😩

[sivar2311]

It might be that there is an interference with SSL, because that's the only main difference i see between our sketches. You can try SinricPro nonSSL Version by putting a #define SINRICPRO_NOSSL before `#include ´

[peastuti]

@sivar2311 that fixed the issue! Thank you a lot! Really nice work! Btw, what does it mean SINRIC_NOSSL? May I have some kind of problem in the future?

[sivar2311]

SINRIC_NOSSL deactivates SSL websocket connection to a standard unsecure websocket connection. So it looks like, that websockets library is not compatible to pubsub library (when SSL is used). Unfortunately this is a thing i cannot fix. Maybe another mqtt library might work here.

Right now, SinricPro supports both (SSL and non SSL connections), but this might change in future.

Edit: It may be that this issue is about SSL. SinricPro SSL only make use of the SSL encryption (without using certificates or fingerprints). Security here is by using app_key / app_secret combination, where app_secret is used to generate HMAC's. Setting a ssl cert

BearSSL::X509List cert(cacert);
BearSSL::X509List client_crt(client_cert);
BearSSL::PrivateKey key(privkey);

might be the problematic part here.

Sorry for all those "may's" and "might's" but i am not an SSL expert - These are only my thoughts on this issue

[peastuti]

Yeah @sivar2311 you are probably right. I'm not an expert on those matters too, I think it could fix the issue if I find a way to connect to MQTT using app key and app secrets too and not using certificates. But, since, as I said, I'm not an expert I haven't found sufficient documentation to do that. Maybe it is not possible using Arduino, or very hard.

[sivar2311]

I am reading right now about BearSSL and certificates. And i think my last edit is correct. AppKey / AppSecret is a thing only SinricPro is using. This is not used by MQTT brokers - so there is no way. But maybe there is another way - but i need to find out more about SSL and certificates for this.

[sivar2311]

The comment in BearSSL_CertStore example says this clearly:

If you know the exact server being connected to, or you are generating your own self-signed certificates and aren't allowing connections to HTTPS/TLS servers out of your control, then you do NOT want a CertStore. Hardcode the self-signing CA or the site's x.509 certificate directly.

"If you know the exact server being connected to..." -> yes we know the server (sinric.pro) "...or you are generating your own self-signed certificates..." -> yes we do (using self signed certificates) "...then you do NOT want CertStore" -> because it won't work (that's what you pointed out)

[sivar2311]

I can't compile the code you posted in your 3rd comment because "secrets.h" and "aquarium_config.h" are missing. I need to know what's inside these files. especially what do you use for cacert and client_cert ? client_cert might be a secret (private) cert generated for your MQTT client?. So question is, what do you use for cacert ? I think this is the point which is causing the issues

[peastuti]

cacert is the CA certificate that I downloaded from AWS IoT Core after creating a new thing, as the tutorial explains. I used this example as reference. As you see it is quite similar. My secrets.h is the same as the link. In aquarium_config I got some configurations about overloading IP addresses of the wifi config, nothing to worry. Is that enough?

[sivar2311]

Ah, yes thank you.

So in conclusion: Setting a cacert will block connecting to SinricPro (because there is no cacert /it is not implemented and used in sdk - yet) Not setting a cacert will block connecting to MQTT server because it needs a cacert.

I guess we identified the problem :)

[peastuti]

Yeah, that's exact. In the meanwhile, I think I could work and study in order to understand if I can use Key/Secret or Username/Password to connect my ESP with AWS MQTT.

[sivar2311]

I think the better way would be to use SINRIC_NOSSL until we get that working. SinricPro messages are secured by HMAC (each message is digitally signed with AppSecret). So keep the SSL secured connection for MQTT

[peastuti]

Yes!! Will do. Thank you a lot for your help!

[sivar2311]

You're welcome, and: thanks for you support 👍

[sivar2311]

Are you still up and willing to do a test? I might have found a solution and need your help (because i dont have SSL secured MQTT broker) Please contact me via sivar2311@googlemail.com

[peastuti]

Yess, just sent you an invitation on hangout

[sivar2311]

Hm, didn't get an invitation (i am not familiar with hangout)

[sivar2311]

Ok, but i think we can do it here via git issues. Can you try this SinricProWebsockets.h please. Don't forget to remove the SINRIC_NOSSL

[sivar2311]

Referring to the analysis by Markus (Links2004): The ESP is unfortunately only able to establish and handle one SSL connection at a time.

The problem here is neither in the PubSubLibrary nor in the SinricPro Library, but simply in the limited capacity of the ESP chip respectively the ESP8266WiFi library.