Eclipse Community Forums: Paho » Not able to simultaneously Publish & Subscribe

Home » Eclipse Projects » Paho » Not able to simultaneously Publish & Subscribe (from client end i'm able to publish data but not able to subscribe in client ..until one condition is not true inside main loop in client(I have two condition check in main loop ))

Show: Today's Messages :: Show Polls :: Message Navigator

Not able to simultaneously Publish & Subscribe [message #1797397]

Tue, 30 October 2018 15:26

vivek kumar

Messages: 6
Registered: October 2018

Junior Member

/*******************************************************************************
* Copyright (c) 2014, 2015 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Eclipse Distribution License v1.0 which accompany this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
* and the Eclipse Distribution License is available at
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* Contributors:
* Ian Craggs - initial contribution
* Benjamin Cabe - adapt to IPStack, and add Yun instructions
* Ian Craggs - remove sprintfs to reduce sketch size
*******************************************************************************/
#define WARN Serial.println

#define MQTTCLIENT_QOS2 1
#include <ESP8266WiFi.h>

#include <IPStack.h>
#include <Countdown.h>
#include <MQTTClient.h>

float state = 0.0 ;
int arrivedcount = 0;

#define WLAN_SSID "xxxxxx"
#define WLAN_PASS "xxxxxxxxxx"

int bulb = 16 ;
byte bulbSwitch = 14;

float bulbState = 0.0;

void messageArrived(MQTT::MessageData& md)
{
MQTT::Message &message = md.message;

Serial.print("Message ");

Serial.print(++arrivedcount);
Serial.print(" arrived: qos ");
Serial.print(message.qos);
Serial.print(", retained ");
Serial.print(message.retained);
Serial.print(", dup ");
Serial.print(message.dup);
Serial.print(", packetid ");
Serial.println(message.id);
Serial.print("Payload ");
Serial.println((char*)message.payload);

if(strcmp((char*)message.payload, "chk") == 0)
{
Serial.println("arrive_chk") ;
//delay(500) ;
chkpub() ;
delay(100) ;

}

else if(strcmp((char*)message.payload, "onn") == 0)
{
Serial.println("arrive_onn") ;
digitalWrite(bulb,HIGH);

}
else if(strcmp((char*)message.payload, "off") == 0)
{
Serial.println("arrive_off") ;
digitalWrite(bulb,LOW);

}

//delay(100) ;
}

WiFiClient Client;
IPStack ipstack(Client);
MQTT::Client<IPStack, Countdown, 50, 1> client = MQTT::Client<IPStack, Countdown, 50, 1>(ipstack);

byte mac[] = { 0x60, 0x01, 0x94, 0x23, 0x7B, 0x58 }; // replace with your device's MAC

const char* topic = "stud/Ligh/cf4b7ba9";

void connect()
{

char hostname[] = "broker.hivemq.com";
int port = 1883;

Serial.print("Connecting to ");
Serial.print(hostname);
Serial.print(":");
Serial.println(port);

int rc = ipstack.connect(hostname, port);
if (rc != 1)
{
Serial.print("rc from TCP connect is ");
//Serial.println(rc);
}

Serial.println("MQTT connecting");
MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
data.MQTTVersion = 3;
data.clientID.cstring = (char*)"stud/Ligh/cf4b7ba9";

rc = client.connect(data);
if (rc != 0)
{
Serial.print("rc from MQTT connect is ");
Serial.println(rc);
}
Serial.println("MQTT connected");

rc = client.subscribe("STUD/LIGH/CF4B7BA9", MQTT::QOS2, messageArrived);
//delay(100) ;
if (rc != 0)
{
Serial.print("rc from MQTT subscribe is ");
Serial.println(rc);

}
Serial.println("MQTT subscribed");

}

void setup()
{
pinMode(bulb, OUTPUT);
pinMode(bulbSwitch, INPUT);
digitalWrite(bulb, LOW) ;
Serial.begin(115200);

Serial.print("Connecting to ");
Serial.println(WLAN_SSID);

WiFi.begin(WLAN_SSID, WLAN_PASS);
while (WiFi.status() != WL_CONNECTED)
{
delay(500);
Serial.print(".");
}
Serial.println();

Serial.println("WiFi connected");
Serial.println("IP address: ");
Serial.println(WiFi.localIP());
connect();
}

MQTT::Message message;

void loop()
{
if (!client.isConnected())
connect();

arrivedcount = 0;

/********************************PROBLEM AREA START****************************************/

Here I am not able to subscribe data until one condition is not true (As I observe... .i am not able to subscribe until I'm not published data.)

Are these conditions blocking to subscribtion ?(Because without condition it is able to do both simultaneously)
/**************************************************************************************************/
if(digitalRead(bulbSwitch)==HIGH && bulbState== 0.0)
{
float x = bulbOnn() ;
bulbPub(x) ;
Serial.println("first_condition") ;
}

else if(digitalRead(bulbSwitch)==LOW && bulbState== 1.0)
{
float y = bulbOff() ;
bulbPub(y) ;
Serial.println("second_condition") ;
}

/************************PROBLEM AREA END****************************************/
}

void bulbPub(float Bulb)
{
Serial.println("bulbPubFunction") ;
char buf[100];
message.payload = (void*)buf;

dtostrf(Bulb,2,1,buf) ;
//Serial.println(buf) ;
message.qos = MQTT::QOS1;
message.payloadlen = strlen(buf)+1;
int rc = client.publish(topic, message);
//Serial.println(rc) ;
delay(100);
}
void chkpub()
{

Serial.println("chkPubFunction") ;
//delay(500) ;

char chk[100];
message.payload = (void*)chk;

if(bulbState == HIGH)
{
dtostrf(1.0,2,1,chk) ;
}
else
{
dtostrf(0.0,2,1,chk) ;
}

//Serial.println(chk) ;
message.qos = MQTT::QOS1;
message.payloadlen = strlen(chk)+1;
int rc = client.publish(topic, message);
delay(100);

}

/*************************on/off function************************/
float bulbOnn()
{
Serial.println("bulbOnnFunction") ;
digitalWrite(bulb, HIGH);
bulbState = 1.0 ;
return bulbState ;
}
float bulbOff()
{
Serial.println("bulbOffFunction") ;
digitalWrite(bulb, LOW);
bulbState = 0.0 ;
return bulbState ;
}

Report message to a moderator

Re: Not able to simultaneously Publish & Subscribe [message #1797400 is a reply to message #1797397]

Tue, 30 October 2018 15:49

Ian Craggs

Messages: 83
Registered: March 2013

Member

This client library is not multi-threaded, so nothing can be received unless you allow it time to do so. Rather than calling delay(), you should preferably use client.yield() which will allow messages to be read from the socket. You should also check that you have subscribed to the correct topic :-)

Report message to a moderator

Now it is working just by publishing random value [message #1797408 is a reply to message #1797400]

Tue, 30 October 2018 16:02

vivek kumar

Messages: 6
Registered: October 2018

Junior Member

/*******************************************************************************
* Copyright (c) 2014, 2015 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Eclipse Distribution License v1.0 which accompany this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
* and the Eclipse Distribution License is available at
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* Contributors:
* Ian Craggs - initial contribution
* Benjamin Cabe - adapt to IPStack, and add Yun instructions
* Ian Craggs - remove sprintfs to reduce sketch size
*******************************************************************************/
#define WARN Serial.println

#define MQTTCLIENT_QOS2 1
#include <ESP8266WiFi.h>

#include <IPStack.h>
#include <Countdown.h>
#include <MQTTClient.h>

float state = 0.0 ;
int arrivedcount = 0;

#define WLAN_SSID "xxxxxx"
#define WLAN_PASS "xxxxxxxxxx"

int bulb = 16 ;
byte bulbSwitch = 14;

float bulbState = 0.0;

void messageArrived(MQTT::MessageData& md)
{
MQTT::Message &message = md.message;

Serial.print("Message ");

Serial.print(++arrivedcount);
Serial.print(" arrived: qos ");
Serial.print(message.qos);
Serial.print(", retained ");
Serial.print(message.retained);
Serial.print(", dup ");
Serial.print(message.dup);
Serial.print(", packetid ");
Serial.println(message.id);
Serial.print("Payload ");
Serial.println((char*)message.payload);

if(strcmp((char*)message.payload, "chk") == 0)
{
Serial.println("arrive_chk") ;
//delay(500) ;
chkpub() ;
delay(100) ;

}

else if(strcmp((char*)message.payload, "onn") == 0)
{
Serial.println("arrive_onn") ;
digitalWrite(bulb,HIGH);

}
else if(strcmp((char*)message.payload, "off") == 0)
{
Serial.println("arrive_off") ;
digitalWrite(bulb,LOW);

}

//delay(100) ;
}

WiFiClient Client;
IPStack ipstack(Client);
MQTT::Client<IPStack, Countdown, 50, 1> client = MQTT::Client<IPStack, Countdown, 50, 1>(ipstack);

byte mac[] = { 0x60, 0x01, 0x94, 0x23, 0x7B, 0x58 }; // replace with your device's MAC

const char* topic = "stud/Ligh/cf4b7ba9";

void connect()
{

char hostname[] = "broker.hivemq.com";
int port = 1883;

Serial.print("Connecting to ");
Serial.print(hostname);
Serial.print(":");
Serial.println(port);

int rc = ipstack.connect(hostname, port);
if (rc != 1)
{
Serial.print("rc from TCP connect is ");
//Serial.println(rc);
}

Serial.println("MQTT connecting");
MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
data.MQTTVersion = 3;
data.clientID.cstring = (char*)"stud/Ligh/cf4b7ba9";

rc = client.connect(data);
if (rc != 0)
{
Serial.print("rc from MQTT connect is ");
Serial.println(rc);
}
Serial.println("MQTT connected");

rc = client.subscribe("STUD/LIGH/CF4B7BA9", MQTT::QOS2, messageArrived);
//delay(100) ;
if (rc != 0)
{
Serial.print("rc from MQTT subscribe is ");
Serial.println(rc);

}
Serial.println("MQTT subscribed");

}

void setup()
{
pinMode(bulb, OUTPUT);
pinMode(bulbSwitch, INPUT);
digitalWrite(bulb, LOW) ;
Serial.begin(115200);

Serial.print("Connecting to ");
Serial.println(WLAN_SSID);

WiFi.begin(WLAN_SSID, WLAN_PASS);
while (WiFi.status() != WL_CONNECTED)
{
delay(500);
Serial.print(".");
}
Serial.println();

Serial.println("WiFi connected");
Serial.println("IP address: ");
Serial.println(WiFi.localIP());
connect();
}

MQTT::Message message;

void loop()
{
if (!client.isConnected())
connect();

arrivedcount = 0;

if(digitalRead(bulbSwitch)==HIGH && bulbState== 0.0)
{
float x = bulbOnn() ;
bulbPub(x) ;
Serial.println("first_condition") ;
}

else if(digitalRead(bulbSwitch)==LOW && bulbState== 1.0)
{
float y = bulbOff() ;
bulbPub(y) ;
Serial.println("second_condition") ;
}

/............MODIFICATION......START 0.................................................................................................../
pub(100.0);
/.................................MODIFICATION......END 0.............................................................................../

}

/............MODIFICATION.......START 1...................................................................................................../
void pub(float Bulb)
{
Serial.println("bulbPubFunction") ;
//delay(500) ;
// Send and receive QoS 1 message
char buf[100];
message.payload = (void*)buf;

dtostrf(Bulb,2,1,buf) ;
//Serial.println(buf) ;
message.qos = MQTT::QOS1;
message.payloadlen = strlen(buf)+1;
int rc = client.publish(topic, message);
//Serial.println(rc) ;
delay(100);
}
/............MODIFICATION.. END...1.............................................................................................../

void bulbPub(float Bulb)
{
Serial.println("bulbPubFunction") ;
char buf[100];
message.payload = (void*)buf;

dtostrf(Bulb,2,1,buf) ;
//Serial.println(buf) ;
message.qos = MQTT::QOS1;
message.payloadlen = strlen(buf)+1;
int rc = client.publish(topic, message);
//Serial.println(rc) ;
delay(100);
}
void chkpub()
{

Serial.println("chkPubFunction") ;
//delay(500) ;

char chk[100];
message.payload = (void*)chk;

if(bulbState == HIGH)
{
dtostrf(1.0,2,1,chk) ;
}
else
{
dtostrf(0.0,2,1,chk) ;
}

//Serial.println(chk) ;
message.qos = MQTT::QOS1;
message.payloadlen = strlen(chk)+1;
int rc = client.publish(topic, message);
delay(100);

}

/*************************on/off function************************/
float bulbOnn()
{
Serial.println("bulbOnnFunction") ;
digitalWrite(bulb, HIGH);
bulbState = 1.0 ;
return bulbState ;
}
float bulbOff()
{
Serial.println("bulbOffFunction") ;
digitalWrite(bulb, LOW);
bulbState = 0.0 ;
return bulbState ;
}

Report message to a moderator