GitHub/paho.mqtt-sn.embedded-c
1
0
Fork 0
mirror of https://github.com/eclipse/paho.mqtt-sn.embedded-c.git synced 2025-12-13 15:36:51 +01:00
Code Issues Projects Releases Wiki Activity

BugFix of #121

Browse Source 
Signed-off-by: tomoaki <tomoaki@tomy-tech.com>
This commit is contained in:
tomoaki
2018-07-18 21:03:03 +09:00
parent a658bd5714
commit 74a9ebaa55
10 changed files with 366 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
MQTTSNGateway/GatewayTester/samples/ClientPub/mainPub.cpp
View File

@@ -77,7 +77,7 @@ MQTTSNCONF = {
const char* topic1 = "ty4tw/topic1"; const char* topic1 = "ty4tw/topic1";
const char* topic2 = "ty4tw/topic2"; const char* topic2 = "ty4tw/topic2";
const char* topic3 = "ty4tw/topic3"; const char* topic3 = "ty4tw/topic3";
const char* topic57 = "ty4tw/topic5/7";
/*------------------------------------------------------ /*------------------------------------------------------
* Callback routines for Subscribed Topics * Callback routines for Subscribed Topics
@@ -112,6 +112,14 @@ void publishTopic2(void)
PUBLISH(topic2,(uint8_t*)payload, strlen(payload), qos); PUBLISH(topic2,(uint8_t*)payload, strlen(payload), qos);
} }
void publishTopic57(void)
{
char payload[300];
sprintf(payload, "publish \"ty4tw/topic57\" \n");
uint8_t qos = 0;
PUBLISH(topic2,(uint8_t*)payload, strlen(payload), qos);
}
void disconnect(void) void disconnect(void)
{ {
@@ -127,7 +135,7 @@ void disconnect(void)
TEST_LIST = {// e.g. TEST( Label, Test), TEST_LIST = {// e.g. TEST( Label, Test),
TEST("Step1:Publish topic1", publishTopic1), TEST("Step1:Publish topic1", publishTopic1),
TEST("Step2:Publish topic2", publishTopic2), TEST("Step2:Publish topic57", publishTopic57),
TEST("Step3:Publish topic2", publishTopic2), TEST("Step3:Publish topic2", publishTopic2),
TEST("Step4:Disconnect", disconnect), TEST("Step4:Disconnect", disconnect),
END_OF_TEST_LIST END_OF_TEST_LIST

23
MQTTSNGateway/GatewayTester/samples/ClientSub/mainSub.cpp
View File

@@ -62,7 +62,7 @@ UDPCONF = {
* Client Configuration (theMqcon) * Client Configuration (theMqcon)
*------------------------------------------------------*/ *------------------------------------------------------*/
MQTTSNCONF = { MQTTSNCONF = {
300, //KeepAlive [seconds] 60, //KeepAlive [seconds]
false, //Clean session false, //Clean session
300, //Sleep duration [seconds] 300, //Sleep duration [seconds]
"", //WillTopic "", //WillTopic
@@ -109,14 +109,22 @@ int on_Topic03(uint8_t* pload, uint16_t ploadlen)
return 0; return 0;
} }
int on_Topic05(uint8_t* pload, uint16_t ploadlen)
{
DISPLAY("\n\nNew callback recv TopicA\n");
pload[ploadlen-1]= 0; // set null terminator
DISPLAY("Payload -->%s　<--\n\n",pload);
return 0;
}
/*------------------------------------------------------ /*------------------------------------------------------
* A Link list of Callback routines and Topics * A Link list of Callback routines and Topics
*------------------------------------------------------*/ *------------------------------------------------------*/
SUBSCRIBE_LIST = {// e.g. SUB(TopicType, topicName, TopicId, callback, QoSx), SUBSCRIBE_LIST = {// e.g. SUB(TopicType, topicName, TopicId, callback, QoSx),
// SUB(MQTTSN_TOPIC_TYPE_NORMAL, topic1, 0, on_Topic01, QoS1), // SUB(MQTTSN_TOPIC_TYPE_NORMAL, topic5, 0, on_Topic05, QoS1),
// SUB(MQTTSN_TOPIC_TYPE_NORMAL, topic2, 0, on_Topic02, QoS1), //SUB(MQTTSN_TOPIC_TYPE_NORMAL, topic2, 0, on_Topic02, QoS1),
END_OF_SUBSCRIBE_LIST END_OF_SUBSCRIBE_LIST
}; };
@@ -139,6 +147,11 @@ void subscribeTopic2(void)
SUBSCRIBE(topic2, on_Topic02, qos); SUBSCRIBE(topic2, on_Topic02, qos);
} }
void subscribeTopic5(void)
{
uint8_t qos = 1;
SUBSCRIBE(topic5, on_Topic05, qos);
}
void disconnect(void) void disconnect(void)
{ {
@@ -162,9 +175,9 @@ void asleep(void)
*------------------------------------------------------*/ *------------------------------------------------------*/
TEST_LIST = {// e.g. TEST( Label, Test), TEST_LIST = {// e.g. TEST( Label, Test),
TEST("Step1:Subscribe topic1", subscribeTopic1), TEST("Step1:Subscribe topic5", subscribeTopic5),
//TEST("Step2:Subscribe topic2", subscribeTopic2), //TEST("Step2:Subscribe topic2", subscribeTopic2),
TEST("Step2:Disconnect", disconnect), TEST("Step2:Disconnect", asleep),
END_OF_TEST_LIST END_OF_TEST_LIST
}; };

5
MQTTSNGateway/GatewayTester/samples/mainTest.cpp
View File

@@ -78,7 +78,10 @@ const char* topic1 = "ty4tw/topic1";
const char* topic2 = "ty4tw/topic2"; const char* topic2 = "ty4tw/topic2";
const char* topic3 = "ty4tw/topic3"; const char* topic3 = "ty4tw/topic3";
const char* topic4 = "ty4tw/topic4"; const char* topic4 = "ty4tw/topic4";
const char* topic5 = "ty4tw/topic5"; const char* topic51 = "ty4tw/topic5/1";
const char* topic52 = "ty4tw/topic5/2";
const char* topic53 = "ty4tw/topic5/3";
const char* topic50 = "ty4tw/topic5/+";
/*------------------------------------------------------ /*------------------------------------------------------

2
MQTTSNGateway/GatewayTester/src/LRegisterManager.cpp
View File

@@ -206,7 +206,7 @@ void LRegisterManager::responceRegister(uint8_t* msg, uint16_t msglen)
uint8_t regack[7]; uint8_t regack[7];
regack[0] = 7; regack[0] = 7;
regack[1] = MQTTSN_TYPE_REGACK; regack[1] = MQTTSN_TYPE_REGACK;
memcpy(regack + 2, msg + 2, 4); memcpy(regack + 2, msg + 1, 4);
LTopic* tp = theClient->getGwProxy()->getTopicTable()->match((char*) msg + 5); LTopic* tp = theClient->getGwProxy()->getTopicTable()->match((char*) msg + 5);
if (tp) if (tp)

545
MQTTSNGateway/GatewayTester/src/LSubscribeManager.cpp
View File

@@ -39,119 +39,119 @@ extern LScreen* theScreen;
=======================================*/ =======================================*/
LSubscribeManager::LSubscribeManager() LSubscribeManager::LSubscribeManager()
{ {
_first = 0; _first = 0;
_last = 0; _last = 0;
} }
LSubscribeManager::~LSubscribeManager() LSubscribeManager::~LSubscribeManager()
{ {
SubElement* elm = _first; SubElement* elm = _first;
SubElement* sav = 0; SubElement* sav = 0;
while (elm) while (elm)
{ {
sav = elm->next; sav = elm->next;
if (elm != 0) if (elm != 0)
{ {
free(elm); free(elm);
} }
elm = sav; elm = sav;
} }
} }
void LSubscribeManager::onConnect(void) void LSubscribeManager::onConnect(void)
{ {
DISPLAY("\033[0m\033[0;32m Attempting OnConnect.....\033[0m\033[0;37m\n"); DISPLAY("\033[0m\033[0;32m Attempting OnConnect.....\033[0m\033[0;37m\n");
if (_first == 0) if (_first == 0)
{ {
for (uint8_t i = 0; theOnPublishList[i].topic != 0; i++) for (uint8_t i = 0; theOnPublishList[i].topic != 0; i++)
{ {
if ( theOnPublishList[i].type == MQTTSN_TOPIC_TYPE_PREDEFINED) if ( theOnPublishList[i].type == MQTTSN_TOPIC_TYPE_PREDEFINED)
{ {
subscribe(theOnPublishList[i].id, theOnPublishList[i].pubCallback, theOnPublishList[i].qos); subscribe(theOnPublishList[i].id, theOnPublishList[i].pubCallback, theOnPublishList[i].qos);
} }
else else
{ {
subscribe(theOnPublishList[i].topic, theOnPublishList[i].pubCallback, theOnPublishList[i].qos); subscribe(theOnPublishList[i].topic, theOnPublishList[i].pubCallback, theOnPublishList[i].qos);
} }
} }
} }
else else
{ {
SubElement* elm = _first; SubElement* elm = _first;
SubElement* pelm; SubElement* pelm;
do do
{ {
pelm = elm; pelm = elm;
if (elm->msgType == MQTTSN_TYPE_SUBSCRIBE) if (elm->msgType == MQTTSN_TYPE_SUBSCRIBE)
{ {
elm->done = SUB_READY; elm->done = SUB_READY;
elm->retryCount = MQTTSN_RETRY_COUNT; elm->retryCount = MQTTSN_RETRY_COUNT;
subscribe(elm->topicName, elm->callback, elm->qos); subscribe(elm->topicName, elm->callback, elm->qos);
} }
elm = pelm->next; elm = pelm->next;
} while (pelm->next); } while (pelm->next);
} }
while (!theClient->getSubscribeManager()->isDone()) while (!theClient->getSubscribeManager()->isDone())
{ {
theClient->getGwProxy()->getMessage(); theClient->getGwProxy()->getMessage();
} }
DISPLAY("\033[0m\033[0;32m OnConnect complete\033[0m\033[0;37m\n"); DISPLAY("\033[0m\033[0;32m OnConnect complete\033[0m\033[0;37m\n");
DISPLAY("\033[0m\033[0;32m Test is Ready.\033[0m\033[0;37m\n"); DISPLAY("\033[0m\033[0;32m Test is Ready.\033[0m\033[0;37m\n");
} }
bool LSubscribeManager::isDone(void) bool LSubscribeManager::isDone(void)
{ {
SubElement* elm = _first; SubElement* elm = _first;
SubElement* prevelm; SubElement* prevelm;
while (elm) while (elm)
{ {
prevelm = elm; prevelm = elm;
if (elm->done == SUB_READY) if (elm->done == SUB_READY)
{ {
return false; return false;
} }
elm = prevelm->next; elm = prevelm->next;
} }
return true; return true;
} }
void LSubscribeManager::send(SubElement* elm) void LSubscribeManager::send(SubElement* elm)
{ {
if (elm->done == SUB_DONE) if (elm->done == SUB_DONE)
{ {
return; return;
} }
uint8_t msg[MQTTSN_MAX_MSG_LENGTH + 1]; uint8_t msg[MQTTSN_MAX_MSG_LENGTH + 1];
if (elm->topicType == MQTTSN_TOPIC_TYPE_NORMAL) if (elm->topicType == MQTTSN_TOPIC_TYPE_PREDEFINED)
{ {
msg[0] = 5 + strlen(elm->topicName); msg[0] = 7;
strcpy((char*) msg + 5, elm->topicName); setUint16(msg + 5, elm->topicId);
} }
else else
{ {
msg[0] = 7; msg[0] = 5 + strlen(elm->topicName);
setUint16(msg + 5, elm->topicId); strcpy((char*) msg + 5, elm->topicName);
} }
msg[1] = elm->msgType; msg[1] = elm->msgType;
msg[2] = elm->qos | elm->topicType; msg[2] = elm->qos | elm->topicType;
if (elm->retryCount == MQTTSN_RETRY_COUNT) if (elm->retryCount == MQTTSN_RETRY_COUNT)
{ {
elm->msgId = theClient->getGwProxy()->getNextMsgId(); elm->msgId = theClient->getGwProxy()->getNextMsgId();
} }
if ((elm->retryCount < MQTTSN_RETRY_COUNT) && elm->msgType == MQTTSN_TYPE_SUBSCRIBE) if ((elm->retryCount < MQTTSN_RETRY_COUNT) && elm->msgType == MQTTSN_TYPE_SUBSCRIBE)
{ {
msg[2] = msg[2] | MQTTSN_FLAG_DUP; msg[2] = msg[2] | MQTTSN_FLAG_DUP;
} }
setUint16(msg + 3, elm->msgId); setUint16(msg + 3, elm->msgId);
theClient->getGwProxy()->connect(); theClient->getGwProxy()->connect();
theClient->getGwProxy()->writeMsg(msg); theClient->getGwProxy()->writeMsg(msg);
theClient->getGwProxy()->setPingReqTimer(); theClient->getGwProxy()->setPingReqTimer();
elm->sendUTC = time(NULL); elm->sendUTC = time(NULL);
elm->retryCount--; elm->retryCount--;
} }
void LSubscribeManager::subscribe(const char* topicName, TopicCallback onPublish, uint8_t qos) void LSubscribeManager::subscribe(const char* topicName, TopicCallback onPublish, uint8_t qos)
@@ -171,8 +171,8 @@ void LSubscribeManager::subscribe(const char* topicName, TopicCallback onPublish
void LSubscribeManager::subscribe(uint16_t topicId, TopicCallback onPublish, uint8_t qos) void LSubscribeManager::subscribe(uint16_t topicId, TopicCallback onPublish, uint8_t qos)
{ {
SubElement* elm = add(MQTTSN_TYPE_SUBSCRIBE, 0, MQTTSN_TOPIC_TYPE_PREDEFINED, topicId, qos, onPublish); SubElement* elm = add(MQTTSN_TYPE_SUBSCRIBE, 0, MQTTSN_TOPIC_TYPE_PREDEFINED, topicId, qos, onPublish);
send(elm); send(elm);
} }
void LSubscribeManager::unsubscribe(const char* topicName) void LSubscribeManager::unsubscribe(const char* topicName)
@@ -186,220 +186,221 @@ void LSubscribeManager::unsubscribe(const char* topicName)
{ {
topicType = MQTTSN_TOPIC_TYPE_SHORT; topicType = MQTTSN_TOPIC_TYPE_SHORT;
} }
SubElement* elm = add(MQTTSN_TYPE_UNSUBSCRIBE, topicName, topicType, 0, 0, 0); SubElement* elm = add(MQTTSN_TYPE_UNSUBSCRIBE, topicName, topicType, 0, 0, 0);
send(elm); send(elm);
} }
void LSubscribeManager::unsubscribe( uint16_t topicId) void LSubscribeManager::unsubscribe( uint16_t topicId)
{ {
SubElement* elm = add(MQTTSN_TYPE_UNSUBSCRIBE, 0, MQTTSN_TOPIC_TYPE_PREDEFINED, topicId, 0, 0); SubElement* elm = add(MQTTSN_TYPE_UNSUBSCRIBE, 0, MQTTSN_TOPIC_TYPE_PREDEFINED, topicId, 0, 0);
send(elm); send(elm);
} }
void LSubscribeManager::checkTimeout(void) void LSubscribeManager::checkTimeout(void)
{ {
SubElement* elm = _first; SubElement* elm = _first;
while (elm) while (elm)
{ {
if (elm->sendUTC + MQTTSN_TIME_RETRY < time(NULL)) if (elm->sendUTC + MQTTSN_TIME_RETRY < time(NULL))
{ {
if (elm->retryCount >= 0) if (elm->retryCount >= 0)
{ {
send(elm); send(elm);
} }
else else
{ {
if ( elm->done == SUB_READY ) if ( elm->done == SUB_READY )
{ {
if (elm->msgType == MQTTSN_TYPE_SUBSCRIBE) if (elm->msgType == MQTTSN_TYPE_SUBSCRIBE)
{ {
DISPLAY("\033[0m\033[0;31m\n!!!!!! SUBSCRIBE Error !!!!! Topic : %s\033[0m\033[0;37m\n\n", (char*)elm->topicName); DISPLAY("\033[0m\033[0;31m\n!!!!!! SUBSCRIBE Error !!!!! Topic : %s\033[0m\033[0;37m\n\n", (char*)elm->topicName);
}else{ }else{
DISPLAY("\033[0m\033[0;31m\n!!!!!! UNSUBSCRIBE Error !!!!! Topic : %s\033[0m\033[0;37m\n\n", (char*)elm->topicName); DISPLAY("\033[0m\033[0;31m\n!!!!!! UNSUBSCRIBE Error !!!!! Topic : %s\033[0m\033[0;37m\n\n", (char*)elm->topicName);
} }
elm->done = SUB_DONE; elm->done = SUB_DONE;
} }
} }
} }
elm = elm->next; elm = elm->next;
} }
} }
void LSubscribeManager::responce(const uint8_t* msg) void LSubscribeManager::responce(const uint8_t* msg)
{ {
if (msg[0] == MQTTSN_TYPE_SUBACK) if (msg[0] == MQTTSN_TYPE_SUBACK)
{ {
uint16_t topicId = getUint16(msg + 2); uint16_t topicId = getUint16(msg + 2);
uint16_t msgId = getUint16(msg + 4); uint16_t msgId = getUint16(msg + 4);
uint8_t rc = msg[6]; uint8_t rc = msg[6];
SubElement* elm = getElement(msgId); SubElement* elm = getElement(msgId);
if (elm) if (elm)
{ {
if ( rc == MQTTSN_RC_ACCEPTED ) if ( rc == MQTTSN_RC_ACCEPTED )
{ {
theClient->getGwProxy()->getTopicTable()->add((char*) elm->topicName, elm->topicType, topicId, elm->callback); theClient->getGwProxy()->getTopicTable()->add((char*) elm->topicName, elm->topicType, topicId, elm->callback);
getElement(msgId)->done = SUB_DONE; getElement(msgId)->done = SUB_DONE;
DISPLAY("\033[0m\033[0;32m Topic \"%s\" Id : %d was Subscribed. \033[0m\033[0;37m\n\n", getElement(msgId)->topicName, topicId); DISPLAY("\033[0m\033[0;32m Topic \"%s\" Id : %d was Subscribed. \033[0m\033[0;37m\n\n", getElement(msgId)->topicName, topicId);
} }
else else
{ {
DISPLAY("\033[0m\033[0;31m SUBACK Invalid messageId. %s\033[0m\033[0;37m\n\n", getElement(msgId)->topicName); DISPLAY("\033[0m\033[0;31m SUBACK Invalid messageId. %s\033[0m\033[0;37m\n\n", getElement(msgId)->topicName);
remove(elm); remove(elm);
} }
} }
} }
else if (msg[0] == MQTTSN_TYPE_UNSUBACK) else if (msg[0] == MQTTSN_TYPE_UNSUBACK)
{ {
uint16_t msgId = getUint16(msg + 1); uint16_t msgId = getUint16(msg + 1);
SubElement* elm = getElement(msgId); SubElement* elm = getElement(msgId);
if (elm) if (elm)
{ {
//theClient->getGwProxy()->getTopicTable()->setCallback(elm->topicName, 0); //theClient->getGwProxy()->getTopicTable()->setCallback(elm->topicName, 0);
DISPLAY("\033[0m\033[0;32m Topic \"%s\" was Unsubscribed. \033[0m\033[0;37m\n\n", getElement(msgId)->topicName); DISPLAY("\033[0m\033[0;32m Topic \"%s\" was Unsubscribed. \033[0m\033[0;37m\n\n", getElement(msgId)->topicName);
remove(elm); remove(elm);
} }
else else
{ {
DISPLAY("\033[0m\033[0;31m UNSUBACK Invalid messageId. \033[0m\033[0;37m\n\n"); DISPLAY("\033[0m\033[0;31m UNSUBACK Invalid messageId. \033[0m\033[0;37m\n\n");
} }
} }
} }
/* SubElement operations */ /* SubElement operations */
SubElement* LSubscribeManager::add(uint8_t msgType, const char* topicName, MQTTSN_topicTypes topicType, uint16_t topicId, SubElement* LSubscribeManager::add(uint8_t msgType, const char* topicName, MQTTSN_topicTypes topicType, uint16_t topicId,
uint8_t qos, TopicCallback callback) uint8_t qos, TopicCallback callback)
{ {
SubElement* elm = 0; SubElement* elm = 0;
if (topicName ) if (topicName )
{ {
elm = getElement(topicName, msgType); elm = getElement(topicName, msgType);
} }
else else
{ {
elm = getElement(topicId, topicType); elm = getElement(topicId, topicType);
} }
if ( elm == 0 )
{
elm = (SubElement*) calloc(1, sizeof(SubElement));
if (elm == 0)
{
return 0;
}
if (_last == 0)
{
_first = elm;
_last = elm;
}
else
{
elm->prev = _last;
_last->next = elm;
_last = elm;
}
}
elm->msgType = msgType; if ( elm == 0 )
elm->callback = callback; {
elm->topicName = topicName; elm = (SubElement*) calloc(1, sizeof(SubElement));
elm->topicId = topicId; if (elm == 0)
elm->topicType = topicType; {
return 0;
}
if (_last == 0)
{
_first = elm;
_last = elm;
}
else
{
elm->prev = _last;
_last->next = elm;
_last = elm;
}
}
if (qos == 1) elm->msgType = msgType;
{ elm->callback = callback;
elm->qos = MQTTSN_FLAG_QOS_1; elm->topicName = topicName;
} elm->topicId = topicId;
else if (qos == 2) elm->topicType = topicType;
{
elm->qos = MQTTSN_FLAG_QOS_2;
}
else
{
elm->qos = MQTTSN_FLAG_QOS_0;
}
elm->msgId = 0;
elm->retryCount = MQTTSN_RETRY_COUNT;
elm->done = SUB_READY;
elm->sendUTC = 0;
return elm; if (qos == 1)
{
elm->qos = MQTTSN_FLAG_QOS_1;
}
else if (qos == 2)
{
elm->qos = MQTTSN_FLAG_QOS_2;
}
else
{
elm->qos = MQTTSN_FLAG_QOS_0;
}
elm->msgId = 0;
elm->retryCount = MQTTSN_RETRY_COUNT;
elm->done = SUB_READY;
elm->sendUTC = 0;
return elm;
} }
void LSubscribeManager::remove(SubElement* elm) void LSubscribeManager::remove(SubElement* elm)
{ {
if (elm) if (elm)
{ {
if (elm->prev == 0) if (elm->prev == 0)
{ {
_first = elm->next; _first = elm->next;
if (elm->next != 0) if (elm->next != 0)
{ {
elm->next->prev = 0; elm->next->prev = 0;
_last = elm->next; _last = elm->next;
} }
free(elm); free(elm);
} }
else else
{ {
if ( elm->next == 0 ) if ( elm->next == 0 )
{ {
_last = elm->prev; _last = elm->prev;
} }
elm->prev->next = elm->next; elm->prev->next = elm->next;
free(elm); free(elm);
} }
} }
} }
SubElement* LSubscribeManager::getElement(uint16_t msgId) SubElement* LSubscribeManager::getElement(uint16_t msgId)
{ {
SubElement* elm = _first; SubElement* elm = _first;
while (elm) while (elm)
{ {
if (elm->msgId == msgId) if (elm->msgId == msgId)
{ {
return elm; return elm;
} }
else else
{ {
elm = elm->next; elm = elm->next;
} }
} }
return 0; return 0;
} }
SubElement* LSubscribeManager::getElement(const char* topicName, uint8_t msgType) SubElement* LSubscribeManager::getElement(const char* topicName, uint8_t msgType)
{ {
SubElement* elm = _first; SubElement* elm = _first;
while (elm) while (elm)
{ {
if ( elm->msgType == msgType && strncmp(elm->topicName, topicName, strlen(topicName)) == 0 ) if ( elm->msgType == msgType && strncmp(elm->topicName, topicName, strlen(topicName)) == 0 )
{ {
return elm; return elm;
} }
else else
{ {
elm = elm->next; elm = elm->next;
} }
} }
return 0; return 0;
} }
SubElement* LSubscribeManager::getElement(uint16_t topicId, MQTTSN_topicTypes topicType) SubElement* LSubscribeManager::getElement(uint16_t topicId, MQTTSN_topicTypes topicType)
{ {
SubElement* elm = _first; SubElement* elm = _first;
while (elm) while (elm)
{ {
if (elm->topicId == topicId && elm->topicType == topicType) if (elm->topicId == topicId && elm->topicType == topicType)
{ {
return elm; return elm;
} }
else else
{ {
elm = elm->next; elm = elm->next;
} }
} }
return 0; return 0;
} }

2
MQTTSNGateway/src/MQTTGWPublishHandler.cpp
View File

@@ -82,7 +82,7 @@ void MQTTGWPublishHandler::handlePublish(Client* client, MQTTGWPacket* packet)
MQTTSN_topicid topicId; MQTTSN_topicid topicId;
uint16_t id = 0; uint16_t id = 0;
if (pub.topiclen == 2) if (pub.topiclen <= 2)
{ {
topicId.type = MQTTSN_TOPIC_TYPE_SHORT; topicId.type = MQTTSN_TOPIC_TYPE_SHORT;
*(topicId.data.short_name) = *pub.topic; *(topicId.data.short_name) = *pub.topic;

31
MQTTSNGateway/src/MQTTSNGWClient.cpp
View File

@@ -400,6 +400,7 @@ Client::Client(bool secure)
_nextClient = 0; _nextClient = 0;
_clientSleepPacketQue.setMaxSize(MAX_SAVED_PUBLISH); _clientSleepPacketQue.setMaxSize(MAX_SAVED_PUBLISH);
_hasPredefTopic = false; _hasPredefTopic = false;
_holdPingRequest = false;
} }
Client::~Client() Client::~Client()
@@ -796,6 +797,21 @@ void Client::setOTAClient(Client* cl)
_otaClient =cl; _otaClient =cl;
} }
void Client::holdPingRequest(void)
{
_holdPingRequest = true;
}
void Client::resetPingRequest(void)
{
_holdPingRequest = false;
}
bool Client::isHoldPringReqest(void)
{
return _holdPingRequest;
}
/*===================================== /*=====================================
Class Topic Class Topic
======================================*/ ======================================*/
@@ -1214,7 +1230,7 @@ TopicIdMapelement* TopicIdMap::getElement(uint16_t msgId)
TopicIdMapelement* TopicIdMap::add(uint16_t msgId, uint16_t topicId, MQTTSN_topicTypes type) TopicIdMapelement* TopicIdMap::add(uint16_t msgId, uint16_t topicId, MQTTSN_topicTypes type)
{ {
if ( _cnt > _maxInflight * 2 || topicId == 0) if ( _cnt > _maxInflight * 2 || ( topicId == 0 && type != MQTTSN_TOPIC_TYPE_SHORT ) )
{ {
return 0; return 0;
} }
@@ -1314,6 +1330,7 @@ WaitREGACKPacketList::WaitREGACKPacketList()
{ {
_first = 0; _first = 0;
_end = 0; _end = 0;
_cnt = 0;
} }
WaitREGACKPacketList::~WaitREGACKPacketList() WaitREGACKPacketList::~WaitREGACKPacketList()
@@ -1346,6 +1363,7 @@ int WaitREGACKPacketList::setPacket(MQTTSNPacket* packet, uint16_t REGACKMsgId)
elm->_prev = _end; elm->_prev = _end;
_end = elm; _end = elm;
} }
_cnt++;
return 1; return 1;
} }
@@ -1387,10 +1405,17 @@ void WaitREGACKPacketList::erase(uint16_t REGACKMsgId)
{ {
p->_next->_prev = p->_prev; p->_next->_prev = p->_prev;
} }
break; _cnt--;
// Do not delete element. Element is deleted after sending to Client. break;
// Do not delete element. Element is deleted after sending to Client.
} }
p = p->_next; p = p->_next;
} }
} }
uint8_t WaitREGACKPacketList::getCount(void)
{
return _cnt;
}

8
MQTTSNGateway/src/MQTTSNGWClient.h
View File

@@ -219,8 +219,10 @@ public:
int setPacket(MQTTSNPacket* packet, uint16_t REGACKMsgId); int setPacket(MQTTSNPacket* packet, uint16_t REGACKMsgId);
MQTTSNPacket* getPacket(uint16_t REGACKMsgId); MQTTSNPacket* getPacket(uint16_t REGACKMsgId);
void erase(uint16_t REGACKMsgId); void erase(uint16_t REGACKMsgId);
uint8_t getCount(void);
private: private:
uint8_t _cnt;
waitREGACKPacket* _first; waitREGACKPacket* _first;
waitREGACKPacket* _end; waitREGACKPacket* _end;
}; };
@@ -298,6 +300,10 @@ public:
bool isSensorNetStable(void); bool isSensorNetStable(void);
bool isWaitWillMsg(void); bool isWaitWillMsg(void);
void holdPingRequest(void);
void resetPingRequest(void);
bool isHoldPringReqest(void);
Client* getNextClient(void); Client* getNextClient(void);
Client* getOTAClient(void); Client* getOTAClient(void);
void setOTAClient(Client* cl); void setOTAClient(Client* cl);
@@ -317,6 +323,8 @@ private:
char* _willTopic; char* _willTopic;
char* _willMsg; char* _willMsg;
bool _holdPingRequest;
Timer _keepAliveTimer; Timer _keepAliveTimer;
uint32_t _keepAliveMsec; uint32_t _keepAliveMsec;

20
MQTTSNGateway/src/MQTTSNGWConnectionHandler.cpp
View File

@@ -270,7 +270,7 @@ void MQTTSNConnectionHandler::handlePingreq(Client* client, MQTTSNPacket* packet
{ {
MQTTGWPacket* msg = 0; MQTTGWPacket* msg = 0;
if ( client->isSleep() || client->isAwake() ) if ( ( client->isSleep() || client->isAwake() ) && client->getClientSleepPacket() )
{ {
while ( ( msg = client->getClientSleepPacket() ) != 0 ) while ( ( msg = client->getClientSleepPacket() ) != 0 )
{ {
@@ -281,12 +281,16 @@ void MQTTSNConnectionHandler::handlePingreq(Client* client, MQTTSNPacket* packet
ev->setBrokerRecvEvent(client, msg); ev->setBrokerRecvEvent(client, msg);
_gateway->getPacketEventQue()->post(ev); _gateway->getPacketEventQue()->post(ev);
} }
client->holdPingRequest();
}
else
{
/* send PINGREQ to the broker */
client->resetPingRequest();
MQTTGWPacket* pingreq = new MQTTGWPacket();
pingreq->setHeader(PINGREQ);
Event* evt = new Event();
evt->setBrokerSendEvent(client, pingreq);
_gateway->getBrokerSendQue()->post(evt);
} }
/* send PINGREQ to the broker */
MQTTGWPacket* pingreq = new MQTTGWPacket();
pingreq->setHeader(PINGREQ);
Event* evt = new Event();
evt->setBrokerSendEvent(client, pingreq);
_gateway->getBrokerSendQue()->post(evt);
} }

11
MQTTSNGateway/src/MQTTSNGWPublishHandler.cpp
View File

@@ -198,6 +198,7 @@ void MQTTSNPublishHandler::handleRegAck( Client* client, MQTTSNPacket* packet)
} }
MQTTSNPacket* regAck = client->getWaitREGACKPacketList()->getPacket(msgId); MQTTSNPacket* regAck = client->getWaitREGACKPacketList()->getPacket(msgId);
if ( regAck != 0 ) if ( regAck != 0 )
{ {
client->getWaitREGACKPacketList()->erase(msgId); client->getWaitREGACKPacketList()->erase(msgId);
@@ -205,6 +206,16 @@ void MQTTSNPublishHandler::handleRegAck( Client* client, MQTTSNPacket* packet)
ev->setClientSendEvent(client, regAck); ev->setClientSendEvent(client, regAck);
_gateway->getClientSendQue()->post(ev); _gateway->getClientSendQue()->post(ev);
} }
if (client->isHoldPringReqest() && client->getWaitREGACKPacketList()->getCount() == 0 )
{
/* send PINGREQ to the broker */
client->resetPingRequest();
MQTTGWPacket* pingreq = new MQTTGWPacket();
pingreq->setHeader(PINGREQ);
Event* evt = new Event();
evt->setBrokerSendEvent(client, pingreq);
_gateway->getBrokerSendQue()->post(evt);
}
} }
} }