Cannot Subscribes new topic after setup MQTT done

[tarzan115]

Hi @monstrenyatko I setup everything in setup functions. and I want to subscribe to topic subtest when receiving message subtest from broker. It was returned Success but when I publish a message to subtest topic. my device cannot receive any data from topic subtest.

I tested subscribes 2 topics while setup MQTT, its success, and my device can receive data from both.

can you help me figure out that issue?

thank you so much.

[monstrenyatko]

Could you please share the code example with me? Probably you missed something. Do you call regularly the yield function?

[tarzan115]

This is my code

I write function subscribeMQTT like your example. and then I use it to subscribe to a new topic when receiving from old topic.

void subscribeMQTT(const char* TOPIC)
{
    MqttClient::Error::type rc = mqtt->subscribe(TOPIC, MqttClient::QOS0, processMessage);
    if (rc != MqttClient::Error::SUCCESS)
        {
        LOG_PRINTFLN("Subscribe error: %i", rc);
        LOG_PRINTFLN("Drop connection");
        mqtt->disconnect();
        return;
    }
}

this function I do subcribe to message from old topic.

void processMessage(MqttClient::MessageData& md) {
    const MqttClient::Message& msg = md.message;
    char payload[msg.payloadLen + 1];
    memcpy(payload, msg.payload, msg.payloadLen);
    payload[msg.payloadLen] = '\0';
    LOG_PRINTFLN(
        "Message arrived: qos %d, retained %d, dup %d, packetid %d, payload:[%s]",
        msg.qos, msg.retained, msg.dup, msg.id, payload
    );

        subscribeMQTT(payload);
}

everything just like your example.

void setup() {
    // Setup hardware serial for logging
    Serial.begin(HW_UART_SPEED);
    while (!Serial);

    // Setup WiFi network
    WiFi.mode(WIFI_STA);
    WiFi.hostname("ESP_" MQTT_ID);
    WiFi.begin("ssid", "passphrase");
    LOG_PRINTFLN("\n");
    LOG_PRINTFLN("Connecting to WiFi");
    while (WiFi.status() != WL_CONNECTED) {
        delay(500);
        LOG_PRINTFLN(".");
    }
    LOG_PRINTFLN("Connected to WiFi");
    LOG_PRINTFLN("IP: %s", WiFi.localIP().toString().c_str());

    // Setup MqttClient
    MqttClient::System *mqttSystem = new System;
    MqttClient::Logger *mqttLogger = new MqttClient::LoggerImpl<HardwareSerial>(Serial);
    MqttClient::Network * mqttNetwork = new MqttClient::NetworkClientImpl<WiFiClient>(network, *mqttSystem);
    //// Make 128 bytes send buffer
    MqttClient::Buffer *mqttSendBuffer = new MqttClient::ArrayBuffer<128>();
    //// Make 128 bytes receive buffer
    MqttClient::Buffer *mqttRecvBuffer = new MqttClient::ArrayBuffer<128>();
    //// Allow up to 2 subscriptions simultaneously
    MqttClient::MessageHandlers *mqttMessageHandlers = new MqttClient::MessageHandlersImpl<2>();
    //// Configure client options
    MqttClient::Options mqttOptions;
    ////// Set command timeout to 10 seconds
    mqttOptions.commandTimeoutMs = 10000;
    //// Make client object
    mqtt = new MqttClient(
        mqttOptions, *mqttLogger, *mqttSystem, *mqttNetwork, *mqttSendBuffer,
        *mqttRecvBuffer, *mqttMessageHandlers
    );

        if (!mqtt->isConnected()) {
        // Close connection if exists
        network.stop();
        // Re-establish TCP connection with MQTT broker
        LOG_PRINTFLN("Connecting");
        network.connect("test.mosquitto.org", 1883);
        if (!network.connected()) {
            LOG_PRINTFLN("Can't establish the TCP connection");
            delay(5000);
            ESP.reset();
        }
        // Start new MQTT connection
        MqttClient::ConnectResult connectResult;
        // Connect
        {
            MQTTPacket_connectData options = MQTTPacket_connectData_initializer;
            options.MQTTVersion = 4;
            options.clientID.cstring = (char*)MQTT_ID;
            options.cleansession = true;
            options.keepAliveInterval = 15; // 15 seconds
            MqttClient::Error::type rc = mqtt->connect(options, connectResult);
            if (rc != MqttClient::Error::SUCCESS) {
                LOG_PRINTFLN("Connection error: %i", rc);
                return;
            }
        }
        subscribeMQTT("FirstTopic");
    }
    }
}
}

void loop()
{
     mqtt->yield(30000L);
]

and other function in example I already add in my code.

[monstrenyatko]

@tarzan115 The root-cause of your problem is the combination of the call to subscribeMQTT(payload) where payload is a temporary array on stack and default implementation of the MqttClient::MessageHandlers interface (see MqttClient::MessageHandlersImpl).

MqttClient::MessageHandlersImpl doesn't copy the topic-string. It just stores the provided pointer. When broker publishes something to the payload topic that memory might be already corrupted on the client side by something else on stack.

I need to add an additional alternative implementation of the MqttClient::MessageHandlers interface to support your scenario with dynamic allocation of the topic-strings before storing the pointer.

[tarzan115]

thank you. I'm waiting for that feature 😄

[monstrenyatko]

@tarzan115 Could you please try my last changes? Take a look on MqttClient::MessageHandlersStaticImpl and MqttClient::MessageHandlersDynamicImpl templates. My suggestion is to use MqttClient::MessageHandlersStaticImpl to avoid dynamic allocations.

[tarzan115]

@monstrenyatko thank you so so much. it was successful. could you please upgrade to Arduino libraries.