pubsub library not thread safe.

[jonasbjurel]

Super good MQTT library - kudos! It appears that the library is not thread safe. Introducing lock protecting buffer may be difficult while supporting all hardware . I propose to solve this by documentation: "The pubsub library is not thread safe, in case several threads use the library following methods need to be protected from concurrent use:

• connect()
• disconnect()
• publish()
• loop()"

[Mallo321123]

I think I have a Similar problem. My Setup: ESP8266 My programm recieves a Json via MQTT every few minutes, and Publishes a Json vie MQTT. The Problem: Every time I publish the Json, it Breaks my other Json. Here Is a Piece of my Code:

for (uint8_t i = 0; i < length_d; i++) {    //Here the Json is still Intact
  Serial.print((char)payload_d[i]);
}
Serial.println("    4");

client.publish("temp/display", mqtt_message, true);     //Publishing a different Json

for (uint8_t i = 0; i < length_d; i++) {    //Here the Json is Broken
  Serial.print((char)payload_d[i]);
}
Serial.println("    5");

And here is the Serial Print:

22:54:21:415 -> {"time":2000,"data":[{"name":"Leistung","value":50,"unit":"W"},{"name":"Durchfluss","value":520,"unit":"L"},{"name":"test","value":234,"unit":"J"}]} 4

22:54:21:429 -> y{"temp_display":22.36099243,"humidity_display":57.86027908}"},{"name":"Durchfluss","value":520,"unit":"L"},{"name":"test","value":234,"unit":"J"}]} 5

Is this a Bug?

[jonasbjurel]

How is payload_d and length_d defined?

[Mallo321123]

Yes, its getting defined by my callback function:

void callback(char* topic, byte* payload, unsigned int length) {
  Serial.print("Message arrived [");
  Serial.print(topic);
  Serial.print("] ");
  for (uint8_t i = 0; i < length; i++) {
    Serial.print((char)payload[i]);
  }
  payload_d = payload;
  length_d = length;
}

[jonasbjurel]

Yes, its getting defined by my callback function:

void callback(char* topic, byte* payload, unsigned int length) {
  Serial.print("Message arrived [");
  Serial.print(topic);
  Serial.print("] ");
  for (uint8_t i = 0; i < length; i++) {
    Serial.print((char)payload[i]);
  }
  payload_d = payload;
  length_d = length;
}

But how/where are they declared?

[jonasbjurel]

Yes, its getting defined by my callback function:

void callback(char* topic, byte* payload, unsigned int length) {
  Serial.print("Message arrived [");
  Serial.print(topic);
  Serial.print("] ");
  for (uint8_t i = 0; i < length; i++) {
    Serial.print((char)payload[i]);
  }
  payload_d = payload;
  length_d = length;
}

But how/where are they declared?

Here is your problem, the callback provides you with a buffer (byte payload) owned by the mqtt library, once you return from the callback, the mqtt will likely free the buffer or re-use it, so once you return that memory space can be used by anyone else. You need to copy the buffer in your callback: payload_d = new char[sizeof(char) length]; memcpy(payload_d, length);

And when you later have consumed the content of payload_d, you need to release it: delete payload_d;

[Mallo321123]

The problem is, that I am consuming The payload_d slowly over time

[jonasbjurel]

The problem is, that I am consuming The payload_d slowly over time

There is no magic, you cannot use the buffer that pubSub will re-circulate or free once you return from the callback, you need to copy it to a buffer you have allocated/own. If you need to retain the information also after the next callback arrives you can do so by introducing an array of buffers which you maintain. But at some point of time you need to free your buffers - or you will exhaust your Heap. That is just reallity. Your problem seems to be unrelated to the lack of thread-safeing which was originally reported in this issue (#1055)

[jonasbjurel]

The problem is, that I am consuming The payload_d slowly over time

There is no magic, you cannot use the buffer that pubSub will re-circulate or free once you return from the callback, you need to copy it to a buffer you have allocated/own. If you need to retain the information also after the next callback arrives you can do so by introducing an array of buffers which you maintain. But at some point of time you need to free your buffers - or you will exhaust your Heap. That is just reallity. Your problem seems to be unrelated to the lack of thread-safeing which was originally reported in this issue (#1055)

Your problem is anyway not related to the problem reported in this thread.