Hello,
I want to receive data from ubidots server. For this I am using a library named pubsubclient. I need two variables of the same topic. I need this variable in the main loop for further work. However, the subscribe function of the pubsubclient only returns a boolean value. How to access the variables those I received from server?
Looking at the excellent PubSubCLient examples would give some clues as to how to do the thing.
Receiving a payload from a MQTT Broker.
subscribe to the topics.
void connectToMQTT()
{
byte mac[5]; // create client ID from mac address
WiFi.macAddress(mac); // get mac address
String clientID = String(mac[0]) + String(mac[4]) ; // use mac address to create clientID
while ( !MQTTclient.connected() )
{
MQTTclient.connect( clientID.c_str(), mqtt_username, mqtt_password );
vTaskDelay( 250 );
}
MQTTclient.setCallback( mqttCallback );
MQTTclient.subscribe ( topicOK );
MQTTclient.subscribe ( topicRemainingMoisture_0 );
MQTTclient.subscribe ( topicWindSpeed );
MQTTclient.subscribe ( topicWindDirection );
MQTTclient.subscribe ( topicRainfall );
MQTTclient.subscribe ( topicWSVolts );
MQTTclient.subscribe ( topicWSCurrent );
MQTTclient.subscribe ( topicWSPower );
MQTTclient.subscribe ( topicDPnWI );
} //void connectToMQTT()
Run client.loop() frequently.
void MQTTkeepalive( void *pvParameters )
{
sema_MQTT_KeepAlive = xSemaphoreCreateBinary();
xSemaphoreGive( sema_MQTT_KeepAlive ); // found keep alive can mess with a publish, stop keep alive during publish
MQTTclient.setKeepAlive( 90 ); // setting keep alive to 90 seconds makes for a very reliable connection, must be set before the 1st connection is made.
TickType_t xLastWakeTime = xTaskGetTickCount();
const TickType_t xFrequency = 250; //delay for ms
for (;;)
{
//check for a is-connected and if the WiFi 'thinks' its connected, found checking on both is more realible than just a single check
if ( (wifiClient.connected()) && (WiFi.status() == WL_CONNECTED) )
{
xSemaphoreTake( sema_MQTT_KeepAlive, portMAX_DELAY ); // whiles MQTTlient.loop() is running no other mqtt operations should be in process
MQTTclient.loop();
xSemaphoreGive( sema_MQTT_KeepAlive );
}
else {
log_i( "MQTT keep alive found MQTT status % s WiFi status % s", String(wifiClient.connected()), String(WiFi.status()) );
if ( !(wifiClient.connected()) || !(WiFi.status() == WL_CONNECTED) )
{
connectToWiFi();
}
connectToMQTT();
}
//log_i( " high watermark % d", uxTaskGetStackHighWaterMark( NULL ) );
xLastWakeTime = xTaskGetTickCount();
vTaskDelayUntil( &xLastWakeTime, xFrequency );
}
vTaskDelete ( NULL );
}
receive the payload in the callback
void IRAM_ATTR mqttCallback(char* topic, byte * payload, unsigned int length)
{
memset( x_message.payload, '\0', payloadSize ); // clear payload char buffer
x_message.topic = ""; //clear topic string buffer
x_message.topic = topic; //store new topic
int i = 0; // extract payload
for ( i; i < length; i++)
{
x_message.payload[i] = ((char)payload[i]);
}
x_message.payload[i] = '\0';
xQueueOverwrite( xQ_Message, (void *) &x_message );// send data to queue
} // void mqttCallback(char* topic, byte* payload, unsigned int length)
parse the received payload.
void fparseMQTT( void *pvParameters )
{
struct stu_message px_message;
for (;;)
{
if ( xQueueReceive(xQ_Message, &px_message, portMAX_DELAY) == pdTRUE )
{
xSemaphoreTake( sema_mqttOK, portMAX_DELAY );
mqttOK = 0;
xSemaphoreGive( sema_mqttOK );
if ( px_message.topic == topicRemainingMoisture_0 )
{
x_eData.RM0 = String(px_message.payload).toFloat();
}
if ( px_message.topic == topicWindSpeed )
{
x_eData.WS = String(px_message.payload).toFloat();
}
if ( px_message.topic == topicWindDirection )
{
x_eData.WD = "";
x_eData.WD = String(px_message.payload);
}
if ( px_message.topic == topicRainfall )
{
x_eData.RF = String(px_message.payload).toFloat();
}
if ( px_message.topic == topicWSVolts )
{
x_eData.WSV = String(px_message.payload).toFloat();
}
if ( px_message.topic == topicWSCurrent )
{
x_eData.WSC = String(px_message.payload).toFloat();
}
if ( px_message.topic == topicWSPower )
{
x_eData.WSP = String(px_message.payload).toFloat();
}
if ( px_message.topic == topicDPnWI )
{
xQueueSend( xQ_WindChillDewPoint, (void *) &px_message, 1 );
}
if ( String(px_message.topic) == topicOK )
{
if ( !TimeSet)
{
String temp = "";
temp.reserve(50);
temp.concat( String(px_message.payload[0]) );
temp.concat( String(px_message.payload[1]) );
temp.concat( String(px_message.payload[2]) );
temp.concat( String(px_message.payload[3]) );
int year = temp.toInt();
temp = "";
temp.concat( String(px_message.payload[5]) + String(px_message.payload[6]) );
int month = temp.toInt();
temp = "";
temp.concat(String(px_message.payload[8]) + String(px_message.payload[9]) );
int day = temp.toInt();
temp = "";
temp.concat( String(px_message.payload[11]) + String(px_message.payload[12]) );
int hour = temp.toInt();
temp = "";
temp.concat( String(px_message.payload[14]) + String(px_message.payload[15]) );
int min = temp.toInt();
rtc.setTime( 0, min, hour, day, month, year );
log_i( "rtc %s Year %d month %d day %d", rtc.getTime(), rtc.getYear(), (rtc.getMonth() + 1), rtc.getDay() );
TimeSet = true;
}
}
} //if ( xQueueReceive(xQ_Message, &px_message, portMAX_DELAY) == pdTRUE )
} //for(;;)
vTaskDelete( NULL );
} // void fparseMQTT( void *pvParameters )