MQTT fails with multiple connections

[Manuel_o]

Hello friends.

Since 3 days i am fighting with this problem and i cant figure out what the problem is.
I am simply collecting my temp. and humidity at 2 different places. To do so, i am using MQTT

The problem: I have 2 MC. Both are sending data to different topics on MQTT.
Both are subscribing to different topics on MQTT ( to make the
MC parameterizable)

When 1 MC is connected, everything is fine. But when both MC are connected everything went downhill and i cant figure out why. It took me 2 days to recognize that somehow they cause the problem when both are connected to LAN. I cant even articulate what the problem my be.

Here is my (simplyfied code)

#define ETH_CLK_MODE ETH_CLOCK_GPIO17_OUT
#define ETH_PHY_POWER 12

#include <ETH.h>
//#include <ESP32Ping.h>
#include <WiFi.h>
#include "time.h"
#include <EEPROM.h>
#include "ESP32_MailClient.h"
#include <Preferences.h>
#include <AsyncTCP.h>
#include <ESPAsyncWebServer.h>
#include <OneWire.h>
#include <DallasTemperature.h>
#include <Smoothed.h> 	// Include the library
#include "globals.h"
#include "DHT.h"
#include <SPI.h>
#include <PubSubClient.h>

//String Company = "Test1";
String Company = "Test2";


#define POWER_SENSE 39
#define BATTERY 35
#define DHTPIN 1
#define DHTTYPE DHT11

// To send Emails using Gmail on port 465 (SSL), you need to create an app password: https://support.google.com/accounts/answer/185833
#define emailSenderAccount    "hello@hello.com"
#define emailSenderPassword   "password"
#define smtpServer            "smtp.office365.com"
#define smtpServerPort        587
#define emailSubject          "[ALERT] "

#define emailSenderAccount_mon    "hello@hello.com"
#define emailSenderPassword_mon   "password"
#define smtpServer_mon            "smtp.office365.com"
#define smtpServerPort_mon        587
#define emailSubject_mon         "[MONITORING] "

//--------------------------------------
//topics send to mqtt
String topic_macadress = Company + "/data/macadress";
String topic_devicename = Company + "/data/devicename";           
String topic_status = Company + "/data/status";           
String topic_timestamp = Company + "/data/timestamp";   
String topic_temperature = Company + "/data/temperature";                       
String topic_huminity = Company + "/data/huminity";                    
String topic_flood = Company + "/data/flood";
String topic_battery = Company + "/data/battery";

//topics listen from mqtt
//-----------------------
//Subscribe to monitoring Email
String topic_mo_header = Company + "/form/mo_header";
String topic_mo_time = Company + "/form/mo_time";           
String topic_mo_rec1 = Company + "/form/mo_rec1";           
String topic_mo_rec2 = Company + "/form/mo_rec2";   
String topic_mo_rec3 = Company + "/form/mo_rec3";                       
String topic_mo_rec4 = Company + "/form/mo_rec4";                    
String topic_mo_onoff = Company + "/form/mo_onoff";
// Subscribe to alert Email;
String topic_al_header = Company + "/form/al_header";
String topic_al_rec1 = Company + "/form/al_rec1";           
String topic_al_rec2 = Company + "/form/al_rec2";           
String topic_al_rec3 = Company + "/form/al_rec3";   
String topic_al_rec4 = Company + "/form/al_rec4";                       
String topic_al_onoff = Company + "/form/al_onoff";
//Subscribe to input  
String topic_t_min = Company + "/form/t_min";
String topic_t_max = Company + "/form/t_max";           
String topic_h_min = Company + "/form/h_min";           
String topic_h_max = Company + "/form/h_max";




/*****Objects*****/
Preferences preferences;

DHT dht(DHTPIN, DHTTYPE);

long lastMsg = 0;
static char macadress[20];
uint32_t statusbuffer;
static char timebuffer[100];
float temperature = 0;
float temperature_last = 0;
float huminity = 0;
float huminity_last = 0;
bool flood = false;
int battery;


/********helpers*****/
static bool eth_connected = false;

// Flag variable to keep track if email notification and SMS notification was sent or not
bool emailSent_temp = false;
bool emailSent_hum = false;
bool smsSent = false;
bool monitoringSent = false;
//---
static int counter = 0;
static int counter_mon = 0;
static int counter_sms = 0;
String emailMessage;


//-----------------T I M E------------------------------
// Variables to save date and time
String formattedDate;
String dayStamp;
String timeStamp;

//struct tm tmS;
//const char* const PROGMEM ntpServer[] = {"fritz.box", "de.pool.ntp.org", "at.pool.ntp.org", "ch.pool.ntp.org", "ptbtime1.ptb.de", "europe.pool.ntp.org"};
const char* ntpServer = "pool.ntp.org";//"at.pool.ntp.org";
const long  gmtOffset_sec = 5;
const int   daylightOffset_sec = 3600;

const char* TZ_INFO    = "CET-1CEST-2,M3.5.0/02:00:00,M10.5.0/03:00:00";
//--------------------------------------------------------


//*******functions*****/
void ReadSensorData_Temp_Hum();
void getTime();
float RandomFloat(float, float);
void checkIfDeviceParameter();
void save_settings_pref(String, String);
String get_settings_pref(String);
void InitialisizeMemberVariabelWith_EPROM_Values();
bool sendEmailNotification(String emailMessage);
void sendCallback(SendStatus msg);
void printLocalTime();
void checkIfDeviceIsOnline();


Smoothed <float> tempSensor; 


class MonitoringEmail {       // The class
  public:             // Access specifier
    String Header;  // Attribute (string variable)
    String Time;
    String Email_I;
    String Email_II;
    String Email_III;
    String Email_IV;
    String OnOff;
};

class AlertEmail {       // The class
  public:             // Access specifier
    String Header;  // Attribute (string variable)
    String Email_I;
    String Email_II;
    String Email_III;
    String Email_IV;
    String OnOff;
};

class Input {       // The class
  public:             // Access specifier
    String temp_min;  // Attribute (string variable)
    String temp_max;
    String hum_min;
    String hum_max;
};



MonitoringEmail monitoringEmail;
AlertEmail alertEmail;
Input input;
// The Email Sending data object contains config and data to send
SMTPData smtpData;








void callback(char* topic, byte* message, unsigned int length) {
  Serial.print("Message arrived on topic: ");
  Serial.print(topic);
  Serial.print(". Message: ");
  String messageTemp;
  
  for (int i = 0; i < length; i++) {
    Serial.print((char)message[i]);
    messageTemp += (char)message[i];
  }
  Serial.println();

  //MonitoringEmail
  if (String(topic) == topic_mo_header) {
    monitoringEmail.Header = messageTemp;
    save_settings_pref("monHeader",monitoringEmail.Header);
  }
  if (String(topic) == topic_mo_time) {
    monitoringEmail.Time = messageTemp;
    save_settings_pref("monTime",monitoringEmail.Time);
  }
  if (String(topic) == topic_mo_rec1) {
      monitoringEmail.Email_I = messageTemp;
      save_settings_pref("monEmail_I",monitoringEmail.Email_I);
  }
  if (String(topic) == topic_mo_rec2) {
      monitoringEmail.Email_II = messageTemp;
      save_settings_pref("monEmail_II",monitoringEmail.Email_II);
  }
  if (String(topic) == topic_mo_rec3) {
      monitoringEmail.Email_III = messageTemp;
      save_settings_pref("monEmail_III",monitoringEmail.Email_III);
  }
  if (String(topic) == topic_mo_rec4) {
      monitoringEmail.Email_IV = messageTemp;
      save_settings_pref("monEmail_IV",monitoringEmail.Email_IV);
  }
  if (String(topic) == topic_mo_onoff) {
      monitoringEmail.OnOff = messageTemp;
      save_settings_pref("monOnOff",monitoringEmail.OnOff);
  }

  //AlertEmail
  if (String(topic) == topic_al_header) {
    alertEmail.Header = messageTemp;
    save_settings_pref("alertHeader",alertEmail.Header);
  }
  if (String(topic) == topic_al_rec1) {
      alertEmail.Email_I = messageTemp;
      save_settings_pref("alertEmail_I",alertEmail.Email_I);
  }
  if (String(topic) == topic_al_rec2) {
      alertEmail.Email_II = messageTemp;
      save_settings_pref("alertEmail_II",alertEmail.Email_II);
  }
  if (String(topic) == topic_al_rec3) {
      alertEmail.Email_III = messageTemp;
      save_settings_pref("alertEmail_III",alertEmail.Email_III);
  }
  if (String(topic) == topic_al_rec4) {
      alertEmail.Email_IV = messageTemp;
      save_settings_pref("alertEmail_IV",alertEmail.Email_IV);
  }
  if (String(topic) == topic_al_onoff) {
      alertEmail.OnOff = messageTemp;
      save_settings_pref("alertOnOff",alertEmail.OnOff);
  }

  //InputFORM_Input/TempMin
    if (String(topic) == topic_t_min) {
    input.temp_min = messageTemp;
    save_settings_pref("inputTMin",input.temp_min);
  }
  if (String(topic) == topic_t_max) {
      input.temp_max = messageTemp;
      save_settings_pref("inputTMax",input.temp_max);
  }
  if (String(topic) == topic_h_min) {
      input.hum_min = messageTemp;
      save_settings_pref("inputHMin",input.hum_min);
  }
  if (String(topic) == topic_h_max) {
       input.hum_max = messageTemp;
      save_settings_pref("inputHMax",input.hum_max);
  }
}


IPAddress server(111, 222, 3, 14);
WiFiClient espClient;
PubSubClient mqttClient(server, 1883, callback, espClient);
long lastReconnectAttempt = 0;
char msg[50];
int value = 0;



void reconnect() {
  // Loop until we're reconnected
  while (!mqttClient.connected()) {
    Serial.print("Attempting MQTT connection...");
    // Attempt to connect
    if (mqttClient.connect("mqtt_ids", "user", "password")) {
      Serial.println("connected");
      // Subscribe to monitoring Email
      mqttClient.subscribe(topic_mo_header.c_str());
      mqttClient.subscribe(topic_mo_time.c_str());
      mqttClient.subscribe(topic_mo_rec1.c_str());
      mqttClient.subscribe(topic_mo_rec2.c_str());
      mqttClient.subscribe(topic_mo_rec3.c_str());
      mqttClient.subscribe(topic_mo_rec4.c_str());
      mqttClient.subscribe(topic_mo_onoff.c_str());
      // Subscribe to alert Email
      mqttClient.subscribe(topic_al_header.c_str());
      mqttClient.subscribe(topic_al_rec1.c_str());
      mqttClient.subscribe(topic_al_rec2.c_str());
      mqttClient.subscribe(topic_al_rec3.c_str());
      mqttClient.subscribe(topic_al_rec4.c_str());
      mqttClient.subscribe(topic_al_onoff.c_str());

      //Subscribe to input 
      mqttClient.subscribe(topic_t_min.c_str());
      mqttClient.subscribe(topic_t_max.c_str());
      mqttClient.subscribe(topic_h_min.c_str());
      mqttClient.subscribe(topic_h_max.c_str());

    } 
    
    else {
      Serial.print("failed, rc=");
      Serial.print(mqttClient.state());
      Serial.println(" try again in 5 seconds");
      // Wait 5 seconds before retrying
      delay(5000);
    }
  }
}





void WiFiEvent(WiFiEvent_t event)
{
  switch (event) {
    case SYSTEM_EVENT_ETH_START:
      Serial.println("ETH Started");
      //set eth hostname here
      ETH.setHostname("esp32-ethernet");
      break;
    case SYSTEM_EVENT_ETH_CONNECTED:
      Serial.println("ETH Connected");
      break;
    case SYSTEM_EVENT_ETH_GOT_IP:
      Serial.print("ETH MAC: ");
      Serial.print(ETH.macAddress());
      Serial.print(", IPv4: ");
      Serial.print(ETH.localIP());
      if (ETH.fullDuplex()) {
        Serial.print(", FULL_DUPLEX");
      }
      Serial.print(", ");
      Serial.print(ETH.linkSpeed());
      Serial.println("Mbps");
      eth_connected = true;
      break;
    case SYSTEM_EVENT_ETH_DISCONNECTED:
      Serial.println("ETH Disconnected");
      eth_connected = false;
      break;
    case SYSTEM_EVENT_ETH_STOP:
      Serial.println("ETH Stopped");
      eth_connected = false;
      break;
    default:
      break;
  }
}

void testClient(const char * host, uint16_t port)
{
  Serial.print("\nconnecting to ");
  Serial.println(host);

  WiFiClient client;
  if (!client.connect(host, port)) {
    Serial.println("connection failed");
    return;
  }
  client.printf("GET / HTTP/1.1\r\nHost: %s\r\n\r\n", host);
  while (client.connected() && !client.available());
  while (client.available()) {
    Serial.write(client.read());
  }

  Serial.println("closing connection\n");
  client.stop();
}



void setup(){

  Serial.begin(115200);
  Serial.println();


  WiFi.onEvent(WiFiEvent);
  Serial.print("Connecting ");
 
  ETH.begin();
  delay(3000);

  pinMode (POWER_SENSE, INPUT);
  pinMode (BATTERY, INPUT);

  tempSensor.begin(SMOOTHED_AVERAGE, 10);	


  configTime(gmtOffset_sec, daylightOffset_sec, ntpServer);

  setenv("TZ", TZ_INFO, 1);

 
  Serial.println();
  Serial.println("WiFi or ETH connected.");

  Serial.println("Use this URL to connect: http://" + ETH.localIP().toString() + "/"); // Print the IP address


  mqttClient.connect("server", "user", "password");

  dht.begin();

  InitialisizeMemberVariabelWith_EPROM_Values();

  delay(1000);


  mqttClient.setCallback(callback); 


  emailSent_temp = false;
  emailSent_hum = false;
  
}



void loop() {


  if (!mqttClient.connected()) {
    reconnect();
  }
  mqttClient.loop();



  long now = millis();

  if (now - lastMsg > 5000) {

    ReadSensorData_Temp_Hum();


    // Convert the temperature to a char array
    char tempString[8];
    dtostrf(temperature, 1, 2, tempString);

    // Convert the huminity to a char array
    char statusbufferString[8];
    dtostrf(statusbuffer, 1, 2, statusbufferString);

    // Convert the huminity to a char array
    char huminityString[8];
    dtostrf(huminity, 1, 2, huminityString);

    //Convert the flood to a char array
    char floodString[8];
    dtostrf(flood, 1, 2, floodString);

    //Convert the flood to a char array
    char batteryString[8];
    dtostrf(battery, 1, 2, batteryString);

    mqttClient.publish(topic_macadress.c_str(),macadress);
    mqttClient.publish(topic_devicename.c_str(),Company.c_str());
    mqttClient.publish(topic_status.c_str(),statusbufferString);
    mqttClient.publish(topic_timestamp.c_str(),timebuffer);
    mqttClient.publish(topic_temperature.c_str(),tempString);
    mqttClient.publish(topic_huminity.c_str(),huminityString);
    mqttClient.publish(topic_flood.c_str(),floodString);
    mqttClient.publish(topic_battery.c_str(),batteryString);


    lastMsg = now;

  }
}


//##################################################################################//
//##################################################################################//
//##################### F U N C T I O N S #########################################//

//#######################################################################//
float RandomFloat(float a, float b) {
    float random = ((float) rand()) / (float) RAND_MAX;
    float diff = b - a;
    float r = random * diff;
    return a + r;
}

//#######################################################################//

void ReadSensorData_Temp_Hum()
{
  delay(2000);
  huminity = dht.readHumidity();
  delay(2000);
  temperature = dht.readTemperature();
  //temperature = RandomFloat(24.5,24.6);

  // Check if any reads failed and exit early (to try again).
  if (isnan(huminity) || isnan(temperature)) {
    Serial.println(F("Failed to read from DHT sensor!"));
    huminity = huminity_last;
    temperature = temperature_last;

    return;
  }

  huminity_last = huminity;
  temperature_last = temperature;
}

//#######################################################################//

void save_settings_pref(String key, String value){

  preferences.begin("input_settings",0);
  preferences.putString(key.c_str(),value);
  preferences.end();
}

//#######################################################################//

void InitialisizeMemberVariabelWith_EPROM_Values()
{
  //monitoring-Email
  monitoringEmail.Header = get_settings_pref("monHeader");
  monitoringEmail.Time = get_settings_pref("monTime");
  monitoringEmail.Email_I = get_settings_pref("monEmail_I");
  monitoringEmail.Email_II = get_settings_pref("monEmail_II");
  monitoringEmail.Email_III = get_settings_pref("monEmail_III");
  monitoringEmail.OnOff = get_settings_pref("monOnOff");
  //Alert-Email
  alertEmail.Header = get_settings_pref("alertHeader");
  alertEmail.Email_I = get_settings_pref("alertEmail_I");
  alertEmail.Email_II = get_settings_pref("alertEmail_II");
  alertEmail.Email_III = get_settings_pref("alertEmail_III");
  alertEmail.OnOff = get_settings_pref("alertOnOff");
  //Input
  input.temp_min = get_settings_pref("inputTMin");
  input.temp_max = get_settings_pref("inputTMax");
  input.hum_min = get_settings_pref("inputHMin");
  input.hum_max = get_settings_pref("inputHMax");
}

//#######################################################################//

String get_settings_pref(String key){

  String returnvL;

  preferences.begin("input_settings",0);
  returnvL = preferences.getString(key.c_str(),"");
  preferences.end();
  return returnvL;
}

It kinda seems, when both MCs are connected, the connection to the MQTT broker fails.
I hope someone has an idea. I gave 1 MC the company-name "Test1" and the other "Test2". So the published and subscribed topics should always be different.

thank you

[gfvalvo]

@Manuel_o

It took me 2 days to recognize that somehow they cause the problem when both are connected to LAN. I cant even articulate what the problem my be.

Did you test your two-connection setup with very simple code and messaging topics first before adding all the other stuff like sensor measurements, email, NTP, handling WiFi events, etc? Or did you just smash it all together and then start testing a large, complex project all at once?

[Manuel_o]

thank you for the answer.

I created one full project and now wanted to add a second device.

One question comes to my mind: This part:

void reconnect() {
  // Loop until we're reconnected
  while (!mqttClient.connected()) {
    Serial.print("Attempting MQTT connection...");
    // Attempt to connect
    if (mqttClient.connect("mqtt_ids", "user", "password")) {
      Serial.println("connected");
      // Subscribe to monitoring Email
      mqttClient.subscribe(topic_mo_header.c_str());
      mqttClient.subscribe(topic_mo_time.c_str());

i read that the first argument of mqttClient.connect() is the ID. Stupid question but must the ID be unique? Beacuse i used "mqtt_ids" on both of my devices?!

Thank you

[gfvalvo]

@Manuel_o

must the ID be unique?

Possibly. Example code I've seen generates a random one for each connection.

@Manuel_o

I created one full project and now wanted to add a second device.

Seriously, if it were my project, I'd strip out all the extraneous clutter and just get a simple multi-connection project to work with deterministic messages published by each processor.

[techypants]

Each connected client must have a unique id. If this is example code for the post, then where is the mqtt_ids being generated at?

if (mqttClient.connect("mqtt_ids", "user", "password")) {

The suspected behavior, is each new client that connects with this code, will knock off the first one, and so on....set your two clients up with unique id's and it won't be an issue.