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

Add The forwarder Encapsulation mesage #27 #69

Browse Source 
Signed-off-by: tomoaki <tomoaki@tomy-tech.com>
This commit is contained in:
tomoaki
2018-07-26 17:42:12 +09:00
parent 4d75351a06
commit 9940aadd4b
28 changed files with 1322 additions and 433 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -220,7 +220,7 @@ TASK_LIST = {// e.g. TASK( task, executing duration in second),
*------------------------------------------------------*/
void setup(void)
{
//SetForwarderMode();
}

647
MQTTSNGateway/GatewayTester/src/LGwProxy.cpp
View File

@@ -1,5 +1,5 @@
/**************************************************************************************
* Copyright (c) 2016, Tomoaki Yamaguchi
* Copyright (c) 2016-2018, Tomoaki Yamaguchi
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
@@ -29,44 +29,47 @@ extern void setUint16(uint8_t* pos, uint16_t val);
extern uint16_t getUint16(const uint8_t* pos);
extern LMqttsnClient* theClient;
extern LScreen* theScreen;
/*=====================================
Class GwProxy
======================================*/
static const char* packet_names[] =
{
"ADVERTISE", "SEARCHGW", "GWINFO", "RESERVED", "CONNECT", "CONNACK",
"WILLTOPICREQ", "WILLTOPIC", "WILLMSGREQ", "WILLMSG", "REGISTER", "REGACK",
"PUBLISH", "PUBACK", "PUBCOMP", "PUBREC", "PUBREL", "RESERVED",
"SUBSCRIBE", "SUBACK", "UNSUBSCRIBE", "UNSUBACK", "PINGREQ", "PINGRESP",
"DISCONNECT", "RESERVED", "WILLTOPICUPD", "WILLTOPICRESP", "WILLMSGUPD",
"WILLMSGRESP"
"ADVERTISE", "SEARCHGW", "GWINFO", "RESERVED", "CONNECT", "CONNACK",
"WILLTOPICREQ", "WILLTOPIC", "WILLMSGREQ", "WILLMSG", "REGISTER", "REGACK",
"PUBLISH", "PUBACK", "PUBCOMP", "PUBREC", "PUBREL", "RESERVED",
"SUBSCRIBE", "SUBACK", "UNSUBSCRIBE", "UNSUBACK", "PINGREQ", "PINGRESP",
"DISCONNECT", "RESERVED", "WILLTOPICUPD", "WILLTOPICRESP", "WILLMSGUPD",
"WILLMSGRESP"
};
LGwProxy::LGwProxy(){
_nextMsgId = 0;
_status = GW_LOST;
_nextMsgId = 0;
_status = GW_LOST;
_gwId = 0;
_willTopic = 0;
_willMsg = 0;
_qosWill = 0;
_retainWill = 0;
_tkeepAlive = MQTTSN_DEFAULT_KEEPALIVE;
_tAdv = MQTTSN_DEFAULT_DURATION;
_cleanSession = 0;
_pingStatus = 0;
_connectRetry = MQTTSN_RETRY_COUNT;
_tSleep = 0;
_tWake = 0;
_initialized = 0;
_willTopic = 0;
_willMsg = 0;
_qosWill = 0;
_retainWill = 0;
_tkeepAlive = MQTTSN_DEFAULT_KEEPALIVE;
_tAdv = MQTTSN_DEFAULT_DURATION;
_cleanSession = 0;
_pingStatus = 0;
_connectRetry = MQTTSN_RETRY_COUNT;
_tSleep = 0;
_tWake = 0;
_initialized = 0;
_isForwarderMode = false;
}
LGwProxy::~LGwProxy(){
_topicTbl.clearTopic();
_topicTbl.clearTopic();
}
void LGwProxy::initialize(LUdpConfig netconf, LMqttsnConfig mqconf){
_network.initialize(netconf);
_clientId = netconf.clientId;
_network.initialize(netconf);
_clientId = netconf.clientId;
_willTopic = mqconf.willTopic;
_willMsg = mqconf.willMsg;
_qosWill = mqconf.willQos;
@@ -77,220 +80,220 @@ void LGwProxy::initialize(LUdpConfig netconf, LMqttsnConfig mqconf){
}
void LGwProxy::connect(){
char* pos;
char* pos;
while (_status != GW_CONNECTED){
pos = _msg;
while (_status != GW_CONNECTED){
pos = _msg;
if (_status == GW_SEND_WILLMSG){
*pos++ = 2 + (uint8_t)strlen(_willMsg);
*pos++ = MQTTSN_TYPE_WILLMSG;
strcpy(pos,_willMsg); // WILLMSG
_status = GW_WAIT_CONNACK;
writeGwMsg();
}else if (_status == GW_SEND_WILLTOPIC){
*pos++ = 3 + (uint8_t)strlen(_willTopic);
*pos++ = MQTTSN_TYPE_WILLTOPIC;
*pos++ = _qosWill | _retainWill;
strcpy(pos,_willTopic); // WILLTOPIC
_status = GW_WAIT_WILLMSGREQ;
writeGwMsg();
}else if (_status == GW_CONNECTING || _status == GW_DISCONNECTED || _status == GW_SLEPT ){
uint8_t clientIdLen = uint8_t(strlen(_clientId) > 23 ? 23 : strlen(_clientId));
*pos++ = 6 + clientIdLen;
*pos++ = MQTTSN_TYPE_CONNECT;
pos++;
if (_cleanSession){
_msg[2] = MQTTSN_FLAG_CLEAN;
}
*pos++ = MQTTSN_PROTOCOL_ID;
setUint16((uint8_t*)pos, _tkeepAlive);
pos += 2;
strncpy(pos, _clientId, clientIdLen);
_msg[ 6 + clientIdLen] = 0;
_status = GW_WAIT_CONNACK;
if ( _willMsg && _willTopic && _status != GW_SLEPT ){
if (strlen(_willMsg) && strlen(_willTopic)){
_msg[2] = _msg[2] | MQTTSN_FLAG_WILL; // CONNECT
_status = GW_WAIT_WILLTOPICREQ;
}
}
writeGwMsg();
_connectRetry = MQTTSN_RETRY_COUNT;
}else if (_status == GW_LOST){
if (_status == GW_SEND_WILLMSG){
*pos++ = 2 + (uint8_t)strlen(_willMsg);
*pos++ = MQTTSN_TYPE_WILLMSG;
strcpy(pos,_willMsg); // WILLMSG
_status = GW_WAIT_CONNACK;
writeGwMsg();
}else if (_status == GW_SEND_WILLTOPIC){
*pos++ = 3 + (uint8_t)strlen(_willTopic);
*pos++ = MQTTSN_TYPE_WILLTOPIC;
*pos++ = _qosWill | _retainWill;
strcpy(pos,_willTopic); // WILLTOPIC
_status = GW_WAIT_WILLMSGREQ;
writeGwMsg();
}else if (_status == GW_CONNECTING || _status == GW_DISCONNECTED || _status == GW_SLEPT ){
uint8_t clientIdLen = uint8_t(strlen(_clientId) > 23 ? 23 : strlen(_clientId));
*pos++ = 6 + clientIdLen;
*pos++ = MQTTSN_TYPE_CONNECT;
pos++;
if (_cleanSession){
_msg[2] = MQTTSN_FLAG_CLEAN;
}
*pos++ = MQTTSN_PROTOCOL_ID;
setUint16((uint8_t*)pos, _tkeepAlive);
pos += 2;
strncpy(pos, _clientId, clientIdLen);
_msg[ 6 + clientIdLen] = 0;
_status = GW_WAIT_CONNACK;
if ( _willMsg && _willTopic && _status != GW_SLEPT ){
if (strlen(_willMsg) && strlen(_willTopic)){
_msg[2] = _msg[2] | MQTTSN_FLAG_WILL; // CONNECT
_status = GW_WAIT_WILLTOPICREQ;
}
}
writeGwMsg();
_connectRetry = MQTTSN_RETRY_COUNT;
}else if (_status == GW_LOST){
*pos++ = 3;
*pos++ = MQTTSN_TYPE_SEARCHGW;
*pos = 0; // SERCHGW
_status = GW_SEARCHING;
writeGwMsg();
*pos++ = 3;
*pos++ = MQTTSN_TYPE_SEARCHGW;
*pos = 0; // SERCHGW
_status = GW_SEARCHING;
writeGwMsg();
}
getConnectResponce();
}
return;
}
getConnectResponce();
}
return;
}
int LGwProxy::getConnectResponce(void){
int len = readMsg();
int len = readMsg();
if (len == 0){
if (_sendUTC + MQTTSN_TIME_RETRY < time(NULL)){
if (_msg[1] == MQTTSN_TYPE_CONNECT)
{
_connectRetry--;
}
if (--_retryCount > 0){
writeMsg((const uint8_t*)_msg); // Not writeGwMsg() : not to reset the counter.
_sendUTC = time(NULL);
}else{
_sendUTC = 0;
if ( _status > GW_SEARCHING && _connectRetry > 0){
_status = GW_CONNECTING;
}else{
_status = GW_LOST;
_gwId = 0;
}
return -1;
}
}
return 0;
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_GWINFO && _status == GW_SEARCHING){
_network.setGwAddress();
if (len == 0){
if (_sendUTC + MQTTSN_TIME_RETRY < time(NULL)){
if (_msg[1] == MQTTSN_TYPE_CONNECT)
{
_connectRetry--;
}
if (--_retryCount > 0){
writeMsg((const uint8_t*)_msg); // Not writeGwMsg() : not to reset the counter.
_sendUTC = time(NULL);
}else{
_sendUTC = 0;
if ( _status > GW_SEARCHING && _connectRetry > 0){
_status = GW_CONNECTING;
}else{
_status = GW_LOST;
_gwId = 0;
}
return -1;
}
}
return 0;
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_GWINFO && _status == GW_SEARCHING){
_network.setGwAddress();
_gwId = _mqttsnMsg[1];
_status = GW_CONNECTING;
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_WILLTOPICREQ && _status == GW_WAIT_WILLTOPICREQ){
_status = GW_SEND_WILLTOPIC;
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_WILLMSGREQ && _status == GW_WAIT_WILLMSGREQ){
_status = GW_SEND_WILLMSG;
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_CONNACK && _status == GW_WAIT_CONNACK){
if (_mqttsnMsg[1] == MQTTSN_RC_ACCEPTED){
_status = GW_CONNECTED;
_connectRetry = MQTTSN_RETRY_COUNT;
setPingReqTimer();
if ( _tSleep ){
_tSleep = 0;
}else{
DISPLAY("\033[0m\033[0;32m\n\n Connected to the Broker\033[0m\033[0;37m\n\n");
_status = GW_CONNECTING;
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_WILLTOPICREQ && _status == GW_WAIT_WILLTOPICREQ){
_status = GW_SEND_WILLTOPIC;
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_WILLMSGREQ && _status == GW_WAIT_WILLMSGREQ){
_status = GW_SEND_WILLMSG;
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_CONNACK && _status == GW_WAIT_CONNACK){
if (_mqttsnMsg[1] == MQTTSN_RC_ACCEPTED){
_status = GW_CONNECTED;
_connectRetry = MQTTSN_RETRY_COUNT;
setPingReqTimer();
if ( _tSleep ){
_tSleep = 0;
}else{
DISPLAY("\033[0m\033[0;32m\n\n Connected to the Broker\033[0m\033[0;37m\n\n");
if ( _cleanSession || _initialized == 1 )
{
_topicTbl.clearTopic();
_initialized = 0;
theClient->onConnect(); // SUBSCRIBEs are conducted
}
}
}else{
_status = GW_CONNECTING;
}
}
return 1;
if ( _cleanSession || _initialized == 1 )
{
_topicTbl.clearTopic();
_initialized = 0;
theClient->onConnect(); // SUBSCRIBEs are conducted
}
}
}else{
_status = GW_CONNECTING;
}
}
return 1;
}
void LGwProxy::reconnect(void){
D_MQTTLOG("...Gateway reconnect\r\n");
_status = GW_DISCONNECTED;
connect();
D_MQTTLOG("...Gateway reconnect\r\n");
_status = GW_DISCONNECTED;
connect();
}
void LGwProxy::disconnect(uint16_t secs){
_tSleep = secs;
_tWake = 0;
_msg[1] = MQTTSN_TYPE_DISCONNECT;
_msg[1] = MQTTSN_TYPE_DISCONNECT;
if (secs){
_msg[0] = 4;
setUint16((uint8_t*) _msg + 2, secs);
_status = GW_SLEEPING;
}else{
_msg[0] = 2;
_keepAliveTimer.stop();
_status = GW_DISCONNECTING;
}
if (secs){
_msg[0] = 4;
setUint16((uint8_t*) _msg + 2, secs);
_status = GW_SLEEPING;
}else{
_msg[0] = 2;
_keepAliveTimer.stop();
_status = GW_DISCONNECTING;
}
_retryCount = MQTTSN_RETRY_COUNT;
writeMsg((const uint8_t*)_msg);
_sendUTC = time(NULL);
_retryCount = MQTTSN_RETRY_COUNT;
writeMsg((const uint8_t*)_msg);
_sendUTC = time(NULL);
while ( _status != GW_DISCONNECTED && _status != GW_SLEPT){
if (getDisconnectResponce() < 0){
_status = GW_LOST;
DISPLAY("\033[0m\033[0;31m\n\n!!!!!! DISCONNECT Error !!!!!\033[0m\033[0;37m \n\n");
return;
}
}
while ( _status != GW_DISCONNECTED && _status != GW_SLEPT){
if (getDisconnectResponce() < 0){
_status = GW_LOST;
DISPLAY("\033[0m\033[0;31m\n\n!!!!!! DISCONNECT Error !!!!!\033[0m\033[0;37m \n\n");
return;
}
}
}
int LGwProxy::getDisconnectResponce(void){
int len = readMsg();
int len = readMsg();
if (len == 0){
if (_sendUTC + MQTTSN_TIME_RETRY < time(NULL)){
if (--_retryCount >= 0){
writeMsg((const uint8_t*)_msg);
_sendUTC = time(NULL);
}else{
_status = GW_LOST;
_gwId = 0;
return -1;
}
}
return 0;
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_DISCONNECT){
if (_status == GW_SLEEPING ){
_status = GW_SLEPT;
uint32_t remain = _keepAliveTimer.getRemain();
theClient->setSleepMode(remain);
if (len == 0){
if (_sendUTC + MQTTSN_TIME_RETRY < time(NULL)){
if (--_retryCount >= 0){
writeMsg((const uint8_t*)_msg);
_sendUTC = time(NULL);
}else{
_status = GW_LOST;
_gwId = 0;
return -1;
}
}
return 0;
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_DISCONNECT){
if (_status == GW_SLEEPING ){
_status = GW_SLEPT;
uint32_t remain = _keepAliveTimer.getRemain();
theClient->setSleepMode(remain);
/* Wake up and starts from this point. */
/* Wake up and starts from this point. */
}else{
_status = GW_DISCONNECTED;
}
}
return 0;
}else{
_status = GW_DISCONNECTED;
}
}
return 0;
}
int LGwProxy::getMessage(void){
int len = readMsg();
if (len < 0){
return len; //error
}
int len = readMsg();
if (len < 0){
return len; //error
}
#ifdef DEBUG_MQTTSN
if (len){
D_MQTTLOG(" recved msgType %x\n", _mqttsnMsg[0]);
}
if (len){
D_MQTTLOG(" recved msgType %x\n", _mqttsnMsg[0]);
}
#endif
if (len == 0){
// Check PINGREQ required
checkPingReq();
if (len == 0){
// Check PINGREQ required
checkPingReq();
// Check ADVERTISE valid
checkAdvertise();
// Check ADVERTISE valid
checkAdvertise();
// Check Timeout of REGISTERs
_regMgr.checkTimeout();
// Check Timeout of REGISTERs
_regMgr.checkTimeout();
// Check Timeout of PUBLISHes,
theClient->getPublishManager()->checkTimeout();
// Check Timeout of PUBLISHes,
theClient->getPublishManager()->checkTimeout();
// Check Timeout of SUBSCRIBEs,
// Check Timeout of SUBSCRIBEs,
theClient->getSubscribeManager()->checkTimeout();
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_PUBLISH){
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_PUBLISH){
theClient->getPublishManager()->published(_mqttsnMsg, len);
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_PUBACK || _mqttsnMsg[0] == MQTTSN_TYPE_PUBCOMP ||
_mqttsnMsg[0] == MQTTSN_TYPE_PUBREC || _mqttsnMsg[0] == MQTTSN_TYPE_PUBREL ){
_mqttsnMsg[0] == MQTTSN_TYPE_PUBREC || _mqttsnMsg[0] == MQTTSN_TYPE_PUBREL ){
theClient->getPublishManager()->responce(_mqttsnMsg, (uint16_t)len);
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_SUBACK || _mqttsnMsg[0] == MQTTSN_TYPE_UNSUBACK){
theClient->getSubscribeManager()->responce(_mqttsnMsg);
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_REGISTER){
_regMgr.responceRegister(_mqttsnMsg, len);
_regMgr.responceRegister(_mqttsnMsg, len);
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_REGACK){
_regMgr.responceRegAck(getUint16(_mqttsnMsg + 3), getUint16(_mqttsnMsg + 1));
@@ -301,26 +304,26 @@ int LGwProxy::getMessage(void){
setPingReqTimer();
if ( _tSleep > 0 ){
_tWake += _tkeepAlive;
if ( _tWake < _tSleep ){
theClient->setSleepMode(_tkeepAlive * 1000UL);
}else{
DISPLAY("\033[0m\033[0;32m\n\n Get back to ACTIVE.\033[0m\033[0;37m\n\n");
_tWake = 0;
connect();
}
_tWake += _tkeepAlive;
if ( _tWake < _tSleep ){
theClient->setSleepMode(_tkeepAlive * 1000UL);
}else{
DISPLAY("\033[0m\033[0;32m\n\n Get back to ACTIVE.\033[0m\033[0;37m\n\n");
_tWake = 0;
connect();
}
}
}
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_DISCONNECT){
_status = GW_LOST;
_gwAliveTimer.stop();
_keepAliveTimer.stop();
_status = GW_LOST;
_gwAliveTimer.stop();
_keepAliveTimer.stop();
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_ADVERTISE){
if (getUint16((const uint8_t*)(_mqttsnMsg + 2)) < 61){
_tAdv = getUint16((const uint8_t*)(_mqttsnMsg + 2)) * 1500;
}else{
_tAdv = getUint16((const uint8_t*)(_mqttsnMsg + 2)) * 1100;
}
if (getUint16((const uint8_t*)(_mqttsnMsg + 2)) < 61){
_tAdv = getUint16((const uint8_t*)(_mqttsnMsg + 2)) * 1500;
}else{
_tAdv = getUint16((const uint8_t*)(_mqttsnMsg + 2)) * 1100;
}
_gwAliveTimer.start(_tAdv);
}
return 0;
@@ -330,165 +333,203 @@ int LGwProxy::getMessage(void){
uint16_t LGwProxy::registerTopic(char* topicName, uint16_t topicId){
uint16_t id = topicId;
uint16_t id = topicId;
if (id == 0){
id = _topicTbl.getTopicId(topicName);
_regMgr.registerTopic(topicName);
_regMgr.registerTopic(topicName);
}
return id;
return id;
}
int LGwProxy::writeMsg(const uint8_t* msg){
uint16_t len;
uint8_t pos;
uint8_t rc;
uint16_t len;
uint8_t pos;
uint8_t rc = 0;
if (msg[0] == 0x01){
len = getUint16(msg + 1);
pos = 2;
}else{
len = msg[0];
pos = 1;
}
if (msg[0] == 0x01){
len = getUint16(msg + 1);
pos = 2;
}else{
len = msg[0];
pos = 1;
}
if (msg[0] == 3 && msg[1] == MQTTSN_TYPE_SEARCHGW){
rc = _network.broadcast(msg,len);
}else{
rc = _network.unicast(msg,len);
}
if (msg[0] == 3 && msg[1] == MQTTSN_TYPE_SEARCHGW){
rc = _network.broadcast(msg,len);
}else
{
if ( _isForwarderMode )
{
// create a forwarder encapsulation message WirelessNodeId is a 4bytes fake data
uint8_t* buf = (uint8_t*)malloc(len + 7);
buf[0] = 7;
buf[1] = MQTTSN_TYPE_ENCAPSULATED;
buf[2] = 1;
buf[3] = 'w';
buf[4] = 'n';
buf[5] = 'I';
buf[6] = 'd';
memcpy(buf + 7, msg, len);
if ( buf)
rc = _network.unicast(buf, len + 7);
free(buf);
DISPLAY(" Encapsulated\n ");
}else{
rc = _network.unicast(msg,len);
}
if (rc > 0){
if ( msg[pos] >= MQTTSN_TYPE_ADVERTISE && msg[pos] <= MQTTSN_TYPE_WILLMSGRESP )
{
DISPLAY(" send %s\n", packet_names[msg[pos]]);
}
return rc;
}
//_status = GW_LOST;
//_gwId = 0;
return rc;
if (rc > 0){
if ( msg[pos] >= MQTTSN_TYPE_ADVERTISE && msg[pos] <= MQTTSN_TYPE_WILLMSGRESP )
{
DISPLAY(" send %s\n", packet_names[msg[pos]]);
}
}
}
return rc;
}
void LGwProxy::writeGwMsg(void){
_retryCount = MQTTSN_RETRY_COUNT;
writeMsg((const uint8_t*)_msg);
_sendUTC = time(NULL);
_retryCount = MQTTSN_RETRY_COUNT;
writeMsg((const uint8_t*)_msg);
_sendUTC = time(NULL);
}
int LGwProxy::readMsg(void){
int len = 0;
_mqttsnMsg = _network.getMessage(&len);
if (len == 0){
return 0;
}
int len = 0;
uint8_t* msg = _network.getMessage(&len);
_mqttsnMsg = msg;
if (_mqttsnMsg[0] == 0x01){
int msgLen = (int) getUint16((const uint8_t*)_mqttsnMsg + 1);
if (len != msgLen){
_mqttsnMsg += 3;
len = msgLen - 3;
}
}else{
_mqttsnMsg += 1;
len -= 1;
}
if ( *_mqttsnMsg >= MQTTSN_TYPE_ADVERTISE && *_mqttsnMsg <= MQTTSN_TYPE_WILLMSGRESP )
{
DISPLAY(" recv %s\n", packet_names[*_mqttsnMsg]);
}
return len;
if (len == 0){
return 0;
}
if (_mqttsnMsg[0] == 0x01){
int msgLen = (int) getUint16((const uint8_t*)_mqttsnMsg + 1);
if (len != msgLen){
_mqttsnMsg += 3;
len = msgLen - 3;
}
}else{
_mqttsnMsg += 1;
len -= 1;
}
if ( *_mqttsnMsg == MQTTSN_TYPE_ENCAPSULATED )
{
int lenEncap = len + 1;
if (msg[lenEncap] == 0x01){
int msgLen = (int) getUint16((const uint8_t*)(msg + lenEncap + 1));
msg += (lenEncap + 3);
len = msgLen - 3;
}else{
msg += (lenEncap + 1);
len = *(msg - 1);
}
_mqttsnMsg = msg;
DISPLAY(" recv encapslated message\n" );
}
if ( *_mqttsnMsg >= MQTTSN_TYPE_ADVERTISE && *_mqttsnMsg <= MQTTSN_TYPE_WILLMSGRESP )
{
DISPLAY(" recv %s\n", packet_names[*_mqttsnMsg]);
}
return len;
}
void LGwProxy::setWillTopic(const char* willTopic, uint8_t qos, bool retain){
_willTopic = willTopic;
_retainWill = _qosWill = 0;
if (qos == 1){
_qosWill = MQTTSN_FLAG_QOS_1;
}else if (qos == 2){
_qosWill = MQTTSN_FLAG_QOS_2;
}
if (retain){
_retainWill = MQTTSN_FLAG_RETAIN;
}
_willTopic = willTopic;
_retainWill = _qosWill = 0;
if (qos == 1){
_qosWill = MQTTSN_FLAG_QOS_1;
}else if (qos == 2){
_qosWill = MQTTSN_FLAG_QOS_2;
}
if (retain){
_retainWill = MQTTSN_FLAG_RETAIN;
}
}
void LGwProxy::setWillMsg(const char* willMsg){
_willMsg = willMsg;
_willMsg = willMsg;
}
void LGwProxy::setCleanSession(bool flg){
if (flg){
_cleanSession = MQTTSN_FLAG_CLEAN;
}else{
_cleanSession = 0;
}
if (flg){
_cleanSession = MQTTSN_FLAG_CLEAN;
}else{
_cleanSession = 0;
}
}
uint16_t LGwProxy::getNextMsgId(void){
_nextMsgId++;
_nextMsgId++;
if (_nextMsgId == 0){
_nextMsgId = 1;
_nextMsgId = 1;
}
return _nextMsgId;
}
void LGwProxy::checkPingReq(void){
uint8_t msg[2];
msg[0] = 0x02;
msg[1] = MQTTSN_TYPE_PINGREQ;
msg[0] = 0x02;
msg[1] = MQTTSN_TYPE_PINGREQ;
if ( (_status == GW_CONNECTED || _status == GW_SLEPT) && isPingReqRequired() && _pingStatus != GW_WAIT_PINGRESP){
_pingStatus = GW_WAIT_PINGRESP;
if ( (_status == GW_CONNECTED || _status == GW_SLEPT) && isPingReqRequired() && _pingStatus != GW_WAIT_PINGRESP){
_pingStatus = GW_WAIT_PINGRESP;
_pingRetryCount = MQTTSN_RETRY_COUNT;
writeMsg((const uint8_t*)msg);
writeMsg((const uint8_t*)msg);
_pingSendUTC = time(NULL);
}else if (_pingStatus == GW_WAIT_PINGRESP){
}else if (_pingStatus == GW_WAIT_PINGRESP){
if (_pingSendUTC + MQTTSN_TIME_RETRY < time(NULL)){
if (--_pingRetryCount > 0){
writeMsg((const uint8_t*)msg);
_pingSendUTC = time(NULL);
}else{
_status = GW_LOST;
_gwId = 0;
if (--_pingRetryCount > 0){
writeMsg((const uint8_t*)msg);
_pingSendUTC = time(NULL);
}else{
_status = GW_LOST;
_gwId = 0;
_pingStatus = 0;
_keepAliveTimer.stop();
D_MQTTLOG(" !!! PINGREQ Timeout\n");
}
}
}
}
}
}
}
void LGwProxy::checkAdvertise(void){
if ( _gwAliveTimer.isTimeUp()){
_status = GW_LOST;
_gwId = 0;
_pingStatus = 0;
_gwAliveTimer.stop();
_keepAliveTimer.stop();
D_MQTTLOG(" !!! ADVERTISE Timeout\n");
}
if ( _gwAliveTimer.isTimeUp()){
_status = GW_LOST;
_gwId = 0;
_pingStatus = 0;
_gwAliveTimer.stop();
_keepAliveTimer.stop();
D_MQTTLOG(" !!! ADVERTISE Timeout\n");
}
}
LTopicTable* LGwProxy::getTopicTable(void){
return &_topicTbl;
return &_topicTbl;
}
LRegisterManager* LGwProxy::getRegisterManager(void){
return &_regMgr;
return &_regMgr;
}
bool LGwProxy::isPingReqRequired(void){
return _keepAliveTimer.isTimeUp(_tkeepAlive * 1000UL);
return _keepAliveTimer.isTimeUp(_tkeepAlive * 1000UL);
}
void LGwProxy::setPingReqTimer(void){
_keepAliveTimer.start(_tkeepAlive * 1000UL);
_keepAliveTimer.start(_tkeepAlive * 1000UL);
}
const char* LGwProxy::getClientId(void) {
return _clientId;
return _clientId;
}
void LGwProxy::setForwarderMode(void)
{
_isForwarderMode = true;
}

108
MQTTSNGateway/GatewayTester/src/LGwProxy.h
View File

@@ -51,63 +51,65 @@ namespace linuxAsyncClient {
=======================================*/
class LGwProxy{
public:
LGwProxy();
~LGwProxy();
LGwProxy();
~LGwProxy();
void initialize(LUdpConfig netconf, LMqttsnConfig mqconf);
void connect(void);
void disconnect(uint16_t sec = 0);
int getMessage(void);
uint16_t registerTopic(char* topic, uint16_t toipcId);
void initialize(LUdpConfig netconf, LMqttsnConfig mqconf);
void connect(void);
void disconnect(uint16_t sec = 0);
int getMessage(void);
uint16_t registerTopic(char* topic, uint16_t toipcId);
void setWillTopic(const char* willTopic, uint8_t qos, bool retain = false);
void setWillMsg(const char* willMsg);
void setCleanSession(bool);
void setKeepAliveDuration(uint16_t duration);
void setAdvertiseDuration(uint16_t duration);
void reconnect(void);
int writeMsg(const uint8_t* msg);
void setPingReqTimer(void);
uint16_t getNextMsgId();
LTopicTable* getTopicTable(void);
LRegisterManager* getRegisterManager(void);
const char* getClientId(void);
void setWillTopic(const char* willTopic, uint8_t qos, bool retain = false);
void setWillMsg(const char* willMsg);
void setCleanSession(bool);
void setKeepAliveDuration(uint16_t duration);
void setAdvertiseDuration(uint16_t duration);
void setForwarderMode(void);
void reconnect(void);
int writeMsg(const uint8_t* msg);
void setPingReqTimer(void);
uint16_t getNextMsgId();
LTopicTable* getTopicTable(void);
LRegisterManager* getRegisterManager(void);
const char* getClientId(void);
private:
int readMsg(void);
void writeGwMsg(void);
void checkPingReq(void);
void checkAdvertise(void);
int getConnectResponce(void);
int getDisconnectResponce(void);
bool isPingReqRequired(void);
int readMsg(void);
void writeGwMsg(void);
void checkPingReq(void);
void checkAdvertise(void);
int getConnectResponce(void);
int getDisconnectResponce(void);
bool isPingReqRequired(void);
LNetwork _network;
uint8_t* _mqttsnMsg;
uint16_t _nextMsgId;
const char* _clientId;
const char* _willTopic;
const char* _willMsg;
uint8_t _cleanSession;
uint8_t _initialized;
uint8_t _retainWill;
uint8_t _qosWill;
uint8_t _gwId;
uint16_t _tkeepAlive;
uint32_t _tAdv;
time_t _sendUTC;
int _retryCount;
int _connectRetry;
uint8_t _status;
time_t _pingSendUTC;
uint8_t _pingRetryCount;
uint8_t _pingStatus;
LRegisterManager _regMgr;
LTopicTable _topicTbl;
LTimer _gwAliveTimer;
LTimer _keepAliveTimer;
uint16_t _tSleep;
uint16_t _tWake;
char _msg[MQTTSN_MAX_MSG_LENGTH + 1];
LNetwork _network;
uint8_t* _mqttsnMsg;
uint16_t _nextMsgId;
const char* _clientId;
const char* _willTopic;
const char* _willMsg;
uint8_t _cleanSession;
uint8_t _initialized;
uint8_t _retainWill;
uint8_t _qosWill;
uint8_t _gwId;
uint16_t _tkeepAlive;
uint32_t _tAdv;
time_t _sendUTC;
int _retryCount;
int _connectRetry;
uint8_t _status;
time_t _pingSendUTC;
uint8_t _pingRetryCount;
uint8_t _pingStatus;
LRegisterManager _regMgr;
LTopicTable _topicTbl;
LTimer _gwAliveTimer;
LTimer _keepAliveTimer;
uint16_t _tSleep;
uint16_t _tWake;
bool _isForwarderMode;
char _msg[MQTTSN_MAX_MSG_LENGTH + 1];
};
} /* end of namespace */

4
MQTTSNGateway/GatewayTester/src/LMqttsnClientApp.h
View File

@@ -1,5 +1,5 @@
/**************************************************************************************
* Copyright (c) 2016, Tomoaki Yamaguchi
* Copyright (c) 2016-2018, Tomoaki Yamaguchi
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
@@ -105,6 +105,7 @@ typedef enum
#define END_OF_SUBSCRIBE_LIST {MQTTSN_TOPIC_TYPE_NORMAL,0,0,0, 0}
#define UDPCONF LUdpConfig theNetcon
#define MQTTSNCONF LMqttsnConfig theMqcon
#define SetForwarderMode theClient->getGwProxy()->setForwarderMode
#ifdef CLIENT_MODE
#define DISPLAY(...)
#define PROMPT(...)
@@ -168,6 +169,7 @@ typedef enum
#define MQTTSN_TYPE_WILLTOPICRESP 0x1B
#define MQTTSN_TYPE_WILLMSGUPD 0x1C
#define MQTTSN_TYPE_WILLMSGRESP 0x1D
#define MQTTSN_TYPE_ENCAPSULATED 0xFE
#define MQTTSN_TOPIC_TYPE 0x03

10
MQTTSNGateway/GatewayTester/src/LNetworkUdp.cpp
View File

@@ -76,12 +76,12 @@ uint8_t* LNetwork::getMessage(int* len){
}else{
*len = _rxDataBuf[0];
}
if(recvLen != *len){
*len = 0;
return 0;
}else{
//if(recvLen != *len){
// *len = 0;
// return 0;
//}else{
return _rxDataBuf;
}
//}
}
}
return 0;