QoS-1 PUBLISH is available #34

BugFix of #69

Signed-off-by: tomoaki <tomoaki@tomy-tech.com>
This commit is contained in:
tomoaki
2018-07-29 16:11:38 +09:00
parent 9940aadd4b
commit 4932d2d0ee
32 changed files with 1274 additions and 348 deletions

View File

@@ -14,7 +14,7 @@
</extensions> </extensions>
</storageModule> </storageModule>
<storageModule moduleId="cdtBuildSystem" version="4.0.0"> <storageModule moduleId="cdtBuildSystem" version="4.0.0">
<configuration artifactName="${ProjName}" buildArtefactType="org.eclipse.cdt.build.core.buildArtefactType.exe" buildProperties="org.eclipse.cdt.build.core.buildArtefactType=org.eclipse.cdt.build.core.buildArtefactType.exe,org.eclipse.cdt.build.core.buildType=org.eclipse.cdt.build.core.buildType.debug" cleanCommand="rm -rf" description="" id="cdt.managedbuild.config.gnu.exe.debug.1685199227" name="Debug" optionalBuildProperties="org.eclipse.cdt.docker.launcher.containerbuild.property.volumes=,org.eclipse.cdt.docker.launcher.containerbuild.property.selectedvolumes=" parent="cdt.managedbuild.config.gnu.exe.debug"> <configuration artifactName="${ProjName}" buildArtefactType="org.eclipse.cdt.build.core.buildArtefactType.exe" buildProperties="org.eclipse.cdt.build.core.buildArtefactType=org.eclipse.cdt.build.core.buildArtefactType.exe,org.eclipse.cdt.build.core.buildType=org.eclipse.cdt.build.core.buildType.debug" cleanCommand="rm -rf" description="" id="cdt.managedbuild.config.gnu.exe.debug.1685199227" name="Debug" optionalBuildProperties="org.eclipse.cdt.docker.launcher.containerbuild.property.selectedvolumes=,org.eclipse.cdt.docker.launcher.containerbuild.property.volumes=" parent="cdt.managedbuild.config.gnu.exe.debug">
<folderInfo id="cdt.managedbuild.config.gnu.exe.debug.1685199227." name="/" resourcePath=""> <folderInfo id="cdt.managedbuild.config.gnu.exe.debug.1685199227." name="/" resourcePath="">
<toolChain id="cdt.managedbuild.toolchain.gnu.exe.debug.102264757" name="Linux GCC" superClass="cdt.managedbuild.toolchain.gnu.exe.debug"> <toolChain id="cdt.managedbuild.toolchain.gnu.exe.debug.102264757" name="Linux GCC" superClass="cdt.managedbuild.toolchain.gnu.exe.debug">
<targetPlatform id="cdt.managedbuild.target.gnu.platform.exe.debug.2137160061" name="Debug Platform" superClass="cdt.managedbuild.target.gnu.platform.exe.debug"/> <targetPlatform id="cdt.managedbuild.target.gnu.platform.exe.debug.2137160061" name="Debug Platform" superClass="cdt.managedbuild.target.gnu.platform.exe.debug"/>
@@ -66,7 +66,7 @@
</toolChain> </toolChain>
</folderInfo> </folderInfo>
<sourceEntries> <sourceEntries>
<entry excluding="MQTTSNGateway/src/linux/xbee|MQTTSNGateway/src/linux/udp6|MQTTSNGateway/src/tests|MQTTSNGateway/GatewayTester|MQTTSNGateway/GatewayTester/samples/mainTest.cpp|MQTTSNGateway/GatewayTester/samples/mainTemplate.cpp|MQTTSNGateway/src/tests/mainTestProcessFramework.cpp|MQTTSNGateway/GatewayTester/samples/mainOTA.cpp|MQTTSNGateway/src/mainLogmonitor.cpp|MQTTSNPacket/test|MQTTSNPacket/samples" flags="VALUE_WORKSPACE_PATH|RESOLVED" kind="sourcePath" name=""/> <entry excluding="MQTTSNGateway/src/linux/udp6|MQTTSNGateway/GatewayTester/samples/mainTest.cpp|MQTTSNPacket/test|MQTTSNPacket/samples|MQTTSNGateway/src/mainLogmonitor.cpp|MQTTSNGateway/src/linux/xbee|MQTTSNGateway/GatewayTester/samples/mainTemplate.cpp|MQTTSNGateway/src/tests|MQTTSNGateway/src/tests/mainTestProcessFramework.cpp|MQTTSNGateway/GatewayTester|ClientPubQoS-1|MQTTSNGateway/GatewayTester/samples/mainOTA.cpp" flags="VALUE_WORKSPACE_PATH|RESOLVED" kind="sourcePath" name=""/>
</sourceEntries> </sourceEntries>
</configuration> </configuration>
</storageModule> </storageModule>
@@ -136,8 +136,8 @@
</toolChain> </toolChain>
</folderInfo> </folderInfo>
<sourceEntries> <sourceEntries>
<entry excluding="MQTTSNGateway/GatewayTester|MQTTSNGateway/GatewayTester/samples/mainTest.cpp|MQTTSNGateway/GatewayTester/samples/mainTemplate.cpp|MQTTSNGateway/GatewayTester/samples/mainOTA.cpp|MQTTSNPacket/src|MQTTSNGateway/src|MQTTSNGateway/src/linux|MQTTSNGateway/src/linux/udp|MQTTSNGateway/src/mainLogmonitor.cpp|MQTTSNGateway/src/linux/xbee|MQTTSNPacket/test|MQTTSNPacket/samples" flags="VALUE_WORKSPACE_PATH|RESOLVED" kind="sourcePath" name=""/> <entry excluding="MQTTSNGateway/src/linux/udp|MQTTSNPacket/test|MQTTSNPacket/samples|MQTTSNGateway/src/linux|MQTTSNGateway/GatewayTester/samples/mainOTA.cpp|MQTTSNGateway/GatewayTester/samples/mainTest.cpp|MQTTSNGateway/src/mainLogmonitor.cpp|MQTTSNGateway/GatewayTester/samples/mainTemplate.cpp|MQTTSNGateway/src/linux/xbee|MQTTSNGateway/src|MQTTSNPacket/src|MQTTSNGateway/GatewayTester|ClientPubQoS-1" flags="VALUE_WORKSPACE_PATH|RESOLVED" kind="sourcePath" name=""/>
<entry excluding="tests|tests/mainTestProcessFramework.cpp|mainLogmonitor.cpp|linux|linux/udp" flags="VALUE_WORKSPACE_PATH" kind="sourcePath" name="MQTTSNGateway/src"/> <entry excluding="mainLogmonitor.cpp|tests|tests/mainTestProcessFramework.cpp|linux|linux/udp" flags="VALUE_WORKSPACE_PATH" kind="sourcePath" name="MQTTSNGateway/src"/>
<entry excluding="xbee|udp6" flags="VALUE_WORKSPACE_PATH" kind="sourcePath" name="MQTTSNGateway/src/linux"/> <entry excluding="xbee|udp6" flags="VALUE_WORKSPACE_PATH" kind="sourcePath" name="MQTTSNGateway/src/linux"/>
<entry flags="VALUE_WORKSPACE_PATH" kind="sourcePath" name="MQTTSNPacket/src"/> <entry flags="VALUE_WORKSPACE_PATH" kind="sourcePath" name="MQTTSNPacket/src"/>
</sourceEntries> </sourceEntries>

View File

@@ -7,9 +7,13 @@ PUBAPPL := mainPub
PRGSUB := MQTT-SNSub PRGSUB := MQTT-SNSub
SUBAPPL := mainSub SUBAPPL := mainSub
PRGQOS := MQTT-SNPubQoS-1
QOSAPPL := mainPubQoS-1
SRCDIR := samples SRCDIR := samples
SRCPUB := ClientPub SRCPUB := ClientPub
SRCSUB := ClientSub SRCSUB := ClientSub
SRCQOS := ClientPubQoS-1
SUBDIR := src SUBDIR := src
CPPSRCS := \ CPPSRCS := \
@@ -46,10 +50,11 @@ DEPS := $(CPPSRCS:%.cpp=$(OUTDIR)/%.d)
PROGPUB := $(OUTDIR)/$(PRGPUB) PROGPUB := $(OUTDIR)/$(PRGPUB)
PROGSUB := $(OUTDIR)/$(PRGSUB) PROGSUB := $(OUTDIR)/$(PRGSUB)
PROGQOS := $(OUTDIR)/$(PRGQOS)
.PHONY: install clean .PHONY: install clean
all: $(PROG) $(PROGPUB) $(PROGSUB) all: $(PROG) $(PROGPUB) $(PROGSUB) $(PROGQOS)
@@ -64,6 +69,9 @@ $(PROGPUB): $(OBJS) $(OUTDIR)/$(SRCDIR)/$(SRCPUB)/$(PUBAPPL).o
$(PROGSUB): $(OBJS) $(OUTDIR)/$(SRCDIR)/$(SRCSUB)/$(SUBAPPL).o $(PROGSUB): $(OBJS) $(OUTDIR)/$(SRCDIR)/$(SRCSUB)/$(SUBAPPL).o
$(CXX) $(LDFLAGS) -o $(PROGSUB) $(OUTDIR)/$(SRCDIR)/$(SRCSUB)/$(SUBAPPL).o $(OBJS) $(LIBS) $(LDADD) $(CXX) $(LDFLAGS) -o $(PROGSUB) $(OUTDIR)/$(SRCDIR)/$(SRCSUB)/$(SUBAPPL).o $(OBJS) $(LIBS) $(LDADD)
$(PROGQOS): $(OBJS) $(OUTDIR)/$(SRCDIR)/$(SRCQOS)/$(QOSAPPL).o
$(CXX) $(LDFLAGS) -o $(PROGQOS) $(OUTDIR)/$(SRCDIR)/$(SRCQOS)/$(QOSAPPL).o $(OBJS) $(LIBS) $(LDADD)
$(OUTDIR)/$(SUBDIR)/%.o:$(SUBDIR)/%.cpp $(OUTDIR)/$(SUBDIR)/%.o:$(SUBDIR)/%.cpp
@if [ ! -e `dirname $@` ]; then mkdir -p `dirname $@`; fi @if [ ! -e `dirname $@` ]; then mkdir -p `dirname $@`; fi
@@ -81,6 +89,10 @@ $(OUTDIR)/$(SRCDIR)/$(SRCSUB)/%.o:$(SRCDIR)/$(SRCSUB)%.cpp
@if [ ! -e `dirname $@` ]; then mkdir -p `dirname $@`; fi @if [ ! -e `dirname $@` ]; then mkdir -p `dirname $@`; fi
$(CXX) $(CXXFLAGS) $(CPPFLAGS) $(INCLUDES) $(DEFS) -o $@ -c -MMD -MP -MF $(@:%.o=%.d) $< $(CXX) $(CXXFLAGS) $(CPPFLAGS) $(INCLUDES) $(DEFS) -o $@ -c -MMD -MP -MF $(@:%.o=%.d) $<
$(OUTDIR)/$(SRCDIR)/$(SRCQOS)/%.o:$(SRCDIR)/$(SRCQOS)%.cpp
@if [ ! -e `dirname $@` ]; then mkdir -p `dirname $@`; fi
$(CXX) $(CXXFLAGS) $(CPPFLAGS) $(INCLUDES) $(DEFS) -o $@ -c -MMD -MP -MF $(@:%.o=%.d) $<
clean: clean:
rm -rf $(OUTDIR) rm -rf $(OUTDIR)
@@ -88,4 +100,5 @@ install:
cp -pf $(PROG) ../../../ cp -pf $(PROG) ../../../
cp -pf $(PROGPUB) ../../../ cp -pf $(PROGPUB) ../../../
cp -pf $(PROGSUB) ../../../ cp -pf $(PROGSUB) ../../../
cp -pf $(PROGQOS) ../../../

View File

@@ -0,0 +1,161 @@
/****************************************************************************
* Copyright (c) 2016, Tomoaki Yamaguchi
*
* 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.
*
*---------------------------------------------------------------------------
*
* MQTT-SN GATEWAY TEST CLIENT
*
* Supported functions.
*
* void PUBLISH ( const char* topicName, uint8_t* payload, uint16_t len, uint8_t qos, bool retain = false );
*
* void PUBLISH ( uint16_t topicId, uint8_t* payload, uint16_t len, uint8_t qos, bool retain = false );
*
* void SUBSCRIBE ( const char* topicName, TopicCallback onPublish, uint8_t qos );
*
* void SUBSCRIBE ( uint16_t topicId, TopicCallback onPublish, uint8_t qos );
*
* void UNSUBSCRIBE ( const char* topicName );
*
* void UNSUBSCRIBE ( uint16_t topicId );
*
* void DISCONNECT ( uint16_t sleepInSecs );
*
* void CONNECT ( void );
*
* void DISPLAY( format, .....); <== instead of printf()
*
*
* Contributors:
* Tomoaki Yamaguchi - initial API and implementation
***************************************************************************/
#include "LMqttsnClientApp.h"
#include "LMqttsnClient.h"
#include "LScreen.h"
using namespace std;
using namespace linuxAsyncClient;
extern LMqttsnClient* theClient;
extern LScreen* theScreen;
/*------------------------------------------------------
* UDP Configuration (theNetcon)
*------------------------------------------------------*/
UDPCONF = {
"GatewayTestPubClient", // ClientId
{225,1,1,1}, // Multicast group IP
1883, // Multicast group Port
20001, // Local PortNo
};
/*------------------------------------------------------
* Client Configuration (theMqcon)
*------------------------------------------------------*/
MQTTSNCONF = {
300, //KeepAlive [seconds]
true, //Clean session
300, //Sleep duration [seconds]
"", //WillTopic
"", //WillMessage
0, //WillQos
false //WillRetain
};
/*------------------------------------------------------
* Define Topics
*------------------------------------------------------*/
/*------------------------------------------------------
* Callback routines for Subscribed Topics
*------------------------------------------------------*/
/*------------------------------------------------------
* A Link list of Callback routines and Topics
*------------------------------------------------------*/
SUBSCRIBE_LIST = {// e.g. SUB(TopicType, topicName, TopicId, callback, QoSx),
END_OF_SUBSCRIBE_LIST
};
/*------------------------------------------------------
* Test functions
*------------------------------------------------------*/
void publishTopic1(void)
{
char payload[300];
sprintf(payload, "publish \"ty4tw/Topic1\" \n");
uint8_t qos = 3;
PUBLISH(1,(uint8_t*)payload, strlen(payload), qos);
}
void publishTopic2(void)
{
char payload[300];
sprintf(payload, "publish \"ty4tw/topic2\" \n");
uint8_t qos = 3;
PUBLISH(2,(uint8_t*)payload, strlen(payload), qos);
}
void publishTopic57(void)
{
char payload[300];
sprintf(payload, "publish \"ty4tw/topic57\" \n");
uint8_t qos = 3;
PUBLISH(5,(uint8_t*)payload, strlen(payload), qos);
}
void disconnect(void)
{
DISCONNECT(0);
}
/*------------------------------------------------------
* A List of Test functions is valid in case of
* line 23 of LMqttsnClientApp.h is commented out.
* //#define CLIENT_MODE
*------------------------------------------------------*/
TEST_LIST = {// e.g. TEST( Label, Test),
TEST("Step1:Publish topic1", publishTopic1),
TEST("Step2:Publish topic57", publishTopic57),
TEST("Step3:Publish topic2", publishTopic2),
END_OF_TEST_LIST
};
/*------------------------------------------------------
* List of tasks is invalid in case of line23 of
* LMqttsnClientApp.h is commented out.
* #define CLIENT_MODE
*------------------------------------------------------*/
TASK_LIST = {// e.g. TASK( task, executing duration in second),
TASK(publishTopic1, 4), // publishTopic1() is executed every 4 seconds
TASK(publishTopic2, 7), // publishTopic2() is executed every 7 seconds
END_OF_TASK_LIST
};
/*------------------------------------------------------
* Initialize function
*------------------------------------------------------*/
void setup(void)
{
SetQoSMinus1Mode(true);
}
/***************** END OF PROGRAM ********************/

View File

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

View File

@@ -30,21 +30,16 @@ extern uint16_t getUint16(const uint8_t* pos);
extern LMqttsnClient* theClient; extern LMqttsnClient* theClient;
extern LScreen* theScreen; extern LScreen* theScreen;
/*===================================== /*=====================================
Class GwProxy Class GwProxy
======================================*/ ======================================*/
static const char* packet_names[] = static const char* packet_names[] = { "ADVERTISE", "SEARCHGW", "GWINFO", "RESERVED", "CONNECT", "CONNACK",
{ "WILLTOPICREQ", "WILLTOPIC", "WILLMSGREQ", "WILLMSG", "REGISTER", "REGACK", "PUBLISH", "PUBACK", "PUBCOMP",
"ADVERTISE", "SEARCHGW", "GWINFO", "RESERVED", "CONNECT", "CONNACK", "PUBREC", "PUBREL", "RESERVED", "SUBSCRIBE", "SUBACK", "UNSUBSCRIBE", "UNSUBACK", "PINGREQ", "PINGRESP",
"WILLTOPICREQ", "WILLTOPIC", "WILLMSGREQ", "WILLMSG", "REGISTER", "REGACK", "DISCONNECT", "RESERVED", "WILLTOPICUPD", "WILLTOPICRESP", "WILLMSGUPD", "WILLMSGRESP" };
"PUBLISH", "PUBACK", "PUBCOMP", "PUBREC", "PUBREL", "RESERVED",
"SUBSCRIBE", "SUBACK", "UNSUBSCRIBE", "UNSUBACK", "PINGREQ", "PINGRESP",
"DISCONNECT", "RESERVED", "WILLTOPICUPD", "WILLTOPICRESP", "WILLMSGUPD",
"WILLMSGRESP"
};
LGwProxy::LGwProxy(){ LGwProxy::LGwProxy()
{
_nextMsgId = 0; _nextMsgId = 0;
_status = GW_LOST; _status = GW_LOST;
_gwId = 0; _gwId = 0;
@@ -61,13 +56,16 @@ LGwProxy::LGwProxy(){
_tWake = 0; _tWake = 0;
_initialized = 0; _initialized = 0;
_isForwarderMode = false; _isForwarderMode = false;
_isQoSMinus1Mode = false;
} }
LGwProxy::~LGwProxy(){ LGwProxy::~LGwProxy()
{
_topicTbl.clearTopic(); _topicTbl.clearTopic();
} }
void LGwProxy::initialize(LUdpConfig netconf, LMqttsnConfig mqconf){ void LGwProxy::initialize(LUdpConfig netconf, LMqttsnConfig mqconf)
{
_network.initialize(netconf); _network.initialize(netconf);
_clientId = netconf.clientId; _clientId = netconf.clientId;
_willTopic = mqconf.willTopic; _willTopic = mqconf.willTopic;
@@ -79,31 +77,54 @@ void LGwProxy::initialize(LUdpConfig netconf, LMqttsnConfig mqconf){
_initialized = 1; _initialized = 1;
} }
void LGwProxy::connect(){ void LGwProxy::connect()
{
char* pos; char* pos;
while (_status != GW_CONNECTED){ while (_status != GW_CONNECTED)
{
pos = _msg; pos = _msg;
if (_status == GW_SEND_WILLMSG){ if (_status == GW_LOST)
{
*pos++ = 3;
*pos++ = MQTTSN_TYPE_SEARCHGW;
*pos = 0; // SERCHGW
_status = GW_SEARCHING;
writeGwMsg();
}
else if (_status == GW_SEND_WILLMSG)
{
*pos++ = 2 + (uint8_t) strlen(_willMsg); *pos++ = 2 + (uint8_t) strlen(_willMsg);
*pos++ = MQTTSN_TYPE_WILLMSG; *pos++ = MQTTSN_TYPE_WILLMSG;
strcpy(pos, _willMsg); // WILLMSG strcpy(pos, _willMsg); // WILLMSG
_status = GW_WAIT_CONNACK; _status = GW_WAIT_CONNACK;
writeGwMsg(); writeGwMsg();
}else if (_status == GW_SEND_WILLTOPIC){ }
else if (_status == GW_SEND_WILLTOPIC)
{
*pos++ = 3 + (uint8_t) strlen(_willTopic); *pos++ = 3 + (uint8_t) strlen(_willTopic);
*pos++ = MQTTSN_TYPE_WILLTOPIC; *pos++ = MQTTSN_TYPE_WILLTOPIC;
*pos++ = _qosWill | _retainWill; *pos++ = _qosWill | _retainWill;
strcpy(pos, _willTopic); // WILLTOPIC strcpy(pos, _willTopic); // WILLTOPIC
_status = GW_WAIT_WILLMSGREQ; _status = GW_WAIT_WILLMSGREQ;
writeGwMsg(); writeGwMsg();
}else if (_status == GW_CONNECTING || _status == GW_DISCONNECTED || _status == GW_SLEPT ){ }
else if (_status == GW_CONNECTING || _status == GW_DISCONNECTED || _status == GW_SLEPT)
{
uint8_t clientIdLen = uint8_t(strlen(_clientId) > 23 ? 23 : strlen(_clientId)); uint8_t clientIdLen = uint8_t(strlen(_clientId) > 23 ? 23 : strlen(_clientId));
if (_isQoSMinus1Mode)
{
_status = GW_CONNECTED;
}
else
{
*pos++ = 6 + clientIdLen; *pos++ = 6 + clientIdLen;
*pos++ = MQTTSN_TYPE_CONNECT; *pos++ = MQTTSN_TYPE_CONNECT;
pos++; pos++;
if (_cleanSession){ if (_cleanSession)
{
_msg[2] = MQTTSN_FLAG_CLEAN; _msg[2] = MQTTSN_FLAG_CLEAN;
} }
*pos++ = MQTTSN_PROTOCOL_ID; *pos++ = MQTTSN_PROTOCOL_ID;
@@ -112,45 +133,49 @@ void LGwProxy::connect(){
strncpy(pos, _clientId, clientIdLen); strncpy(pos, _clientId, clientIdLen);
_msg[6 + clientIdLen] = 0; _msg[6 + clientIdLen] = 0;
_status = GW_WAIT_CONNACK; _status = GW_WAIT_CONNACK;
if ( _willMsg && _willTopic && _status != GW_SLEPT ){ if (_willMsg && _willTopic && _status != GW_SLEPT)
if (strlen(_willMsg) && strlen(_willTopic)){ {
if (strlen(_willMsg) && strlen(_willTopic))
{
_msg[2] = _msg[2] | MQTTSN_FLAG_WILL; // CONNECT _msg[2] = _msg[2] | MQTTSN_FLAG_WILL; // CONNECT
_status = GW_WAIT_WILLTOPICREQ; _status = GW_WAIT_WILLTOPICREQ;
} }
} }
writeGwMsg(); writeGwMsg();
_connectRetry = MQTTSN_RETRY_COUNT; _connectRetry = MQTTSN_RETRY_COUNT;
}else if (_status == GW_LOST){ }
*pos++ = 3;
*pos++ = MQTTSN_TYPE_SEARCHGW;
*pos = 0; // SERCHGW
_status = GW_SEARCHING;
writeGwMsg();
} }
getConnectResponce(); getConnectResponce();
} }
return; return;
} }
int LGwProxy::getConnectResponce(void){ int LGwProxy::getConnectResponce(void)
{
int len = readMsg(); int len = readMsg();
if (len == 0){ if (len == 0)
if (_sendUTC + MQTTSN_TIME_RETRY < time(NULL)){ {
if (_sendUTC + MQTTSN_TIME_RETRY < time(NULL))
{
if (_msg[1] == MQTTSN_TYPE_CONNECT) if (_msg[1] == MQTTSN_TYPE_CONNECT)
{ {
_connectRetry--; _connectRetry--;
} }
if (--_retryCount > 0){ if (--_retryCount > 0)
{
writeMsg((const uint8_t*) _msg); // Not writeGwMsg() : not to reset the counter. writeMsg((const uint8_t*) _msg); // Not writeGwMsg() : not to reset the counter.
_sendUTC = time(NULL); _sendUTC = time(NULL);
}else{ }
else
{
_sendUTC = 0; _sendUTC = 0;
if ( _status > GW_SEARCHING && _connectRetry > 0){ if (_status > GW_SEARCHING && _connectRetry > 0)
{
_status = GW_CONNECTING; _status = GW_CONNECTING;
}else{ }
else
{
_status = GW_LOST; _status = GW_LOST;
_gwId = 0; _gwId = 0;
} }
@@ -158,22 +183,34 @@ int LGwProxy::getConnectResponce(void){
} }
} }
return 0; return 0;
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_GWINFO && _status == GW_SEARCHING){ }
else if (_mqttsnMsg[0] == MQTTSN_TYPE_GWINFO && _status == GW_SEARCHING)
{
_network.setGwAddress(); _network.setGwAddress();
_gwId = _mqttsnMsg[1]; _gwId = _mqttsnMsg[1];
_status = GW_CONNECTING; _status = GW_CONNECTING;
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_WILLTOPICREQ && _status == GW_WAIT_WILLTOPICREQ){ }
else if (_mqttsnMsg[0] == MQTTSN_TYPE_WILLTOPICREQ && _status == GW_WAIT_WILLTOPICREQ)
{
_status = GW_SEND_WILLTOPIC; _status = GW_SEND_WILLTOPIC;
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_WILLMSGREQ && _status == GW_WAIT_WILLMSGREQ){ }
else if (_mqttsnMsg[0] == MQTTSN_TYPE_WILLMSGREQ && _status == GW_WAIT_WILLMSGREQ)
{
_status = GW_SEND_WILLMSG; _status = GW_SEND_WILLMSG;
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_CONNACK && _status == GW_WAIT_CONNACK){ }
if (_mqttsnMsg[1] == MQTTSN_RC_ACCEPTED){ else if (_mqttsnMsg[0] == MQTTSN_TYPE_CONNACK && _status == GW_WAIT_CONNACK)
{
if (_mqttsnMsg[1] == MQTTSN_RC_ACCEPTED)
{
_status = GW_CONNECTED; _status = GW_CONNECTED;
_connectRetry = MQTTSN_RETRY_COUNT; _connectRetry = MQTTSN_RETRY_COUNT;
setPingReqTimer(); setPingReqTimer();
if ( _tSleep ){ if (_tSleep)
{
_tSleep = 0; _tSleep = 0;
}else{ }
else
{
DISPLAY("\033[0m\033[0;32m\n\n Connected to the Broker\033[0m\033[0;37m\n\n"); DISPLAY("\033[0m\033[0;32m\n\n Connected to the Broker\033[0m\033[0;37m\n\n");
if (_cleanSession || _initialized == 1) if (_cleanSession || _initialized == 1)
@@ -183,30 +220,37 @@ int LGwProxy::getConnectResponce(void){
theClient->onConnect(); // SUBSCRIBEs are conducted theClient->onConnect(); // SUBSCRIBEs are conducted
} }
} }
}else{ }
else
{
_status = GW_CONNECTING; _status = GW_CONNECTING;
} }
} }
return 1; return 1;
} }
void LGwProxy::reconnect(void){ void LGwProxy::reconnect(void)
{
D_MQTTLOG("...Gateway reconnect\r\n"); D_MQTTLOG("...Gateway reconnect\r\n");
_status = GW_DISCONNECTED; _status = GW_DISCONNECTED;
connect(); connect();
} }
void LGwProxy::disconnect(uint16_t secs){ void LGwProxy::disconnect(uint16_t secs)
{
_tSleep = secs; _tSleep = secs;
_tWake = 0; _tWake = 0;
_msg[1] = MQTTSN_TYPE_DISCONNECT; _msg[1] = MQTTSN_TYPE_DISCONNECT;
if (secs){ if (secs)
{
_msg[0] = 4; _msg[0] = 4;
setUint16((uint8_t*) _msg + 2, secs); setUint16((uint8_t*) _msg + 2, secs);
_status = GW_SLEEPING; _status = GW_SLEEPING;
}else{ }
else
{
_msg[0] = 2; _msg[0] = 2;
_keepAliveTimer.stop(); _keepAliveTimer.stop();
_status = GW_DISCONNECTING; _status = GW_DISCONNECTING;
@@ -216,8 +260,10 @@ void LGwProxy::disconnect(uint16_t secs){
writeMsg((const uint8_t*) _msg); writeMsg((const uint8_t*) _msg);
_sendUTC = time(NULL); _sendUTC = time(NULL);
while ( _status != GW_DISCONNECTED && _status != GW_SLEPT){ while (_status != GW_DISCONNECTED && _status != GW_SLEPT)
if (getDisconnectResponce() < 0){ {
if (getDisconnectResponce() < 0)
{
_status = GW_LOST; _status = GW_LOST;
DISPLAY("\033[0m\033[0;31m\n\n!!!!!! DISCONNECT Error !!!!!\033[0m\033[0;37m \n\n"); DISPLAY("\033[0m\033[0;31m\n\n!!!!!! DISCONNECT Error !!!!!\033[0m\033[0;37m \n\n");
return; return;
@@ -225,48 +271,63 @@ void LGwProxy::disconnect(uint16_t secs){
} }
} }
int LGwProxy::getDisconnectResponce(void){ int LGwProxy::getDisconnectResponce(void)
{
int len = readMsg(); int len = readMsg();
if (len == 0){ if (len == 0)
if (_sendUTC + MQTTSN_TIME_RETRY < time(NULL)){ {
if (--_retryCount >= 0){ if (_sendUTC + MQTTSN_TIME_RETRY < time(NULL))
{
if (--_retryCount >= 0)
{
writeMsg((const uint8_t*) _msg); writeMsg((const uint8_t*) _msg);
_sendUTC = time(NULL); _sendUTC = time(NULL);
}else{ }
else
{
_status = GW_LOST; _status = GW_LOST;
_gwId = 0; _gwId = 0;
return -1; return -1;
} }
} }
return 0; return 0;
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_DISCONNECT){ }
if (_status == GW_SLEEPING ){ else if (_mqttsnMsg[0] == MQTTSN_TYPE_DISCONNECT)
{
if (_status == GW_SLEEPING)
{
_status = GW_SLEPT; _status = GW_SLEPT;
uint32_t remain = _keepAliveTimer.getRemain(); uint32_t remain = _keepAliveTimer.getRemain();
theClient->setSleepMode(remain); theClient->setSleepMode(remain);
/* Wake up and starts from this point. */ /* Wake up and starts from this point. */
}else{ }
else
{
_status = GW_DISCONNECTED; _status = GW_DISCONNECTED;
} }
} }
return 0; return 0;
} }
int LGwProxy::getMessage(void){ int LGwProxy::getMessage(void)
{
int len = readMsg(); int len = readMsg();
if (len < 0){ if (len < 0)
{
return len; //error return len; //error
} }
#ifdef DEBUG_MQTTSN #ifdef DEBUG_MQTTSN
if (len){ if (len)
{
D_MQTTLOG(" recved msgType %x\n", _mqttsnMsg[0]); D_MQTTLOG(" recved msgType %x\n", _mqttsnMsg[0]);
} }
#endif #endif
if (len == 0){ if (len == 0)
{
// Check PINGREQ required // Check PINGREQ required
checkPingReq(); checkPingReq();
@@ -282,46 +343,70 @@ int LGwProxy::getMessage(void){
// Check Timeout of SUBSCRIBEs, // Check Timeout of SUBSCRIBEs,
theClient->getSubscribeManager()->checkTimeout(); theClient->getSubscribeManager()->checkTimeout();
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_PUBLISH){ }
else if (_mqttsnMsg[0] == MQTTSN_TYPE_PUBLISH)
{
theClient->getPublishManager()->published(_mqttsnMsg, len); 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 ){ else if (_mqttsnMsg[0] == MQTTSN_TYPE_PUBACK || _mqttsnMsg[0] == MQTTSN_TYPE_PUBCOMP
|| _mqttsnMsg[0] == MQTTSN_TYPE_PUBREC || _mqttsnMsg[0] == MQTTSN_TYPE_PUBREL)
{
theClient->getPublishManager()->responce(_mqttsnMsg, (uint16_t) len); theClient->getPublishManager()->responce(_mqttsnMsg, (uint16_t) len);
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_SUBACK || _mqttsnMsg[0] == MQTTSN_TYPE_UNSUBACK){ }
else if (_mqttsnMsg[0] == MQTTSN_TYPE_SUBACK || _mqttsnMsg[0] == MQTTSN_TYPE_UNSUBACK)
{
theClient->getSubscribeManager()->responce(_mqttsnMsg); theClient->getSubscribeManager()->responce(_mqttsnMsg);
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_REGISTER){ }
else if (_mqttsnMsg[0] == MQTTSN_TYPE_REGISTER)
{
_regMgr.responceRegister(_mqttsnMsg, len); _regMgr.responceRegister(_mqttsnMsg, len);
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_REGACK){ }
else if (_mqttsnMsg[0] == MQTTSN_TYPE_REGACK)
{
_regMgr.responceRegAck(getUint16(_mqttsnMsg + 3), getUint16(_mqttsnMsg + 1)); _regMgr.responceRegAck(getUint16(_mqttsnMsg + 3), getUint16(_mqttsnMsg + 1));
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_PINGRESP){ }
if (_pingStatus == GW_WAIT_PINGRESP){ else if (_mqttsnMsg[0] == MQTTSN_TYPE_PINGRESP)
{
if (_pingStatus == GW_WAIT_PINGRESP)
{
_pingStatus = 0; _pingStatus = 0;
setPingReqTimer(); setPingReqTimer();
if ( _tSleep > 0 ){ if (_tSleep > 0)
{
_tWake += _tkeepAlive; _tWake += _tkeepAlive;
if ( _tWake < _tSleep ){ if (_tWake < _tSleep)
{
theClient->setSleepMode(_tkeepAlive * 1000UL); theClient->setSleepMode(_tkeepAlive * 1000UL);
}else{ }
else
{
DISPLAY("\033[0m\033[0;32m\n\n Get back to ACTIVE.\033[0m\033[0;37m\n\n"); DISPLAY("\033[0m\033[0;32m\n\n Get back to ACTIVE.\033[0m\033[0;37m\n\n");
_tWake = 0; _tWake = 0;
connect(); connect();
} }
} }
} }
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_DISCONNECT){ }
else if (_mqttsnMsg[0] == MQTTSN_TYPE_DISCONNECT)
{
_status = GW_LOST; _status = GW_LOST;
_gwAliveTimer.stop(); _gwAliveTimer.stop();
_keepAliveTimer.stop(); _keepAliveTimer.stop();
}else if (_mqttsnMsg[0] == MQTTSN_TYPE_ADVERTISE){ }
if (getUint16((const uint8_t*)(_mqttsnMsg + 2)) < 61){ else if (_mqttsnMsg[0] == MQTTSN_TYPE_ADVERTISE)
{
if (getUint16((const uint8_t*) (_mqttsnMsg + 2)) < 61)
{
_tAdv = getUint16((const uint8_t*) (_mqttsnMsg + 2)) * 1500; _tAdv = getUint16((const uint8_t*) (_mqttsnMsg + 2)) * 1500;
}else{ }
else
{
_tAdv = getUint16((const uint8_t*) (_mqttsnMsg + 2)) * 1100; _tAdv = getUint16((const uint8_t*) (_mqttsnMsg + 2)) * 1100;
} }
_gwAliveTimer.start(_tAdv); _gwAliveTimer.start(_tAdv);
@@ -329,35 +414,39 @@ int LGwProxy::getMessage(void){
return 0; return 0;
} }
uint16_t LGwProxy::registerTopic(char* topicName, uint16_t topicId)
{
uint16_t LGwProxy::registerTopic(char* topicName, uint16_t topicId){
uint16_t id = topicId; uint16_t id = topicId;
if (id == 0){ if (id == 0)
{
id = _topicTbl.getTopicId(topicName); id = _topicTbl.getTopicId(topicName);
_regMgr.registerTopic(topicName); _regMgr.registerTopic(topicName);
} }
return id; return id;
} }
int LGwProxy::writeMsg(const uint8_t* msg)
int LGwProxy::writeMsg(const uint8_t* msg){ {
uint16_t len; uint16_t len;
uint8_t pos; uint8_t pos;
uint8_t rc = 0; uint8_t rc = 0;
if (msg[0] == 0x01){ if (msg[0] == 0x01)
{
len = getUint16(msg + 1); len = getUint16(msg + 1);
pos = 2; pos = 2;
}else{ }
else
{
len = msg[0]; len = msg[0];
pos = 1; pos = 1;
} }
if (msg[0] == 3 && msg[1] == MQTTSN_TYPE_SEARCHGW){ if (msg[0] == 3 && msg[1] == MQTTSN_TYPE_SEARCHGW)
{
rc = _network.broadcast(msg, len); rc = _network.broadcast(msg, len);
}else }
else
{ {
if (_isForwarderMode) if (_isForwarderMode)
{ {
@@ -375,11 +464,14 @@ int LGwProxy::writeMsg(const uint8_t* msg){
rc = _network.unicast(buf, len + 7); rc = _network.unicast(buf, len + 7);
free(buf); free(buf);
DISPLAY(" Encapsulated\n "); DISPLAY(" Encapsulated\n ");
}else{ }
else
{
rc = _network.unicast(msg, len); rc = _network.unicast(msg, len);
} }
if (rc > 0){ if (rc > 0)
{
if (msg[pos] >= MQTTSN_TYPE_ADVERTISE && msg[pos] <= MQTTSN_TYPE_WILLMSGRESP) if (msg[pos] >= MQTTSN_TYPE_ADVERTISE && msg[pos] <= MQTTSN_TYPE_WILLMSGRESP)
{ {
DISPLAY(" send %s\n", packet_names[msg[pos]]); DISPLAY(" send %s\n", packet_names[msg[pos]]);
@@ -389,28 +481,35 @@ int LGwProxy::writeMsg(const uint8_t* msg){
return rc; return rc;
} }
void LGwProxy::writeGwMsg(void){ void LGwProxy::writeGwMsg(void)
{
_retryCount = MQTTSN_RETRY_COUNT; _retryCount = MQTTSN_RETRY_COUNT;
writeMsg((const uint8_t*) _msg); writeMsg((const uint8_t*) _msg);
_sendUTC = time(NULL); _sendUTC = time(NULL);
} }
int LGwProxy::readMsg(void){ int LGwProxy::readMsg(void)
{
int len = 0; int len = 0;
uint8_t* msg = _network.getMessage(&len); uint8_t* msg = _network.getMessage(&len);
_mqttsnMsg = msg; _mqttsnMsg = msg;
if (len == 0){ if (len == 0)
{
return 0; return 0;
} }
if (_mqttsnMsg[0] == 0x01){ if (_mqttsnMsg[0] == 0x01)
{
int msgLen = (int) getUint16((const uint8_t*) _mqttsnMsg + 1); int msgLen = (int) getUint16((const uint8_t*) _mqttsnMsg + 1);
if (len != msgLen){ if (len != msgLen)
{
_mqttsnMsg += 3; _mqttsnMsg += 3;
len = msgLen - 3; len = msgLen - 3;
} }
}else{ }
else
{
_mqttsnMsg += 1; _mqttsnMsg += 1;
len -= 1; len -= 1;
} }
@@ -419,11 +518,14 @@ int LGwProxy::readMsg(void){
{ {
int lenEncap = len + 1; int lenEncap = len + 1;
if (msg[lenEncap] == 0x01){ if (msg[lenEncap] == 0x01)
{
int msgLen = (int) getUint16((const uint8_t*) (msg + lenEncap + 1)); int msgLen = (int) getUint16((const uint8_t*) (msg + lenEncap + 1));
msg += (lenEncap + 3); msg += (lenEncap + 3);
len = msgLen - 3; len = msgLen - 3;
}else{ }
else
{
msg += (lenEncap + 1); msg += (lenEncap + 1);
len = *(msg - 1); len = *(msg - 1);
} }
@@ -438,56 +540,75 @@ int LGwProxy::readMsg(void){
return len; return len;
} }
void LGwProxy::setWillTopic(const char* willTopic, uint8_t qos, bool retain){ void LGwProxy::setWillTopic(const char* willTopic, uint8_t qos, bool retain)
{
_willTopic = willTopic; _willTopic = willTopic;
_retainWill = _qosWill = 0; _retainWill = _qosWill = 0;
if (qos == 1){ if (qos == 1)
{
_qosWill = MQTTSN_FLAG_QOS_1; _qosWill = MQTTSN_FLAG_QOS_1;
}else if (qos == 2){ }
else if (qos == 2)
{
_qosWill = MQTTSN_FLAG_QOS_2; _qosWill = MQTTSN_FLAG_QOS_2;
} }
if (retain){ if (retain)
{
_retainWill = MQTTSN_FLAG_RETAIN; _retainWill = MQTTSN_FLAG_RETAIN;
} }
} }
void LGwProxy::setWillMsg(const char* willMsg){ void LGwProxy::setWillMsg(const char* willMsg)
{
_willMsg = willMsg; _willMsg = willMsg;
} }
void LGwProxy::setCleanSession(bool flg)
void LGwProxy::setCleanSession(bool flg){ {
if (flg){ if (flg)
{
_cleanSession = MQTTSN_FLAG_CLEAN; _cleanSession = MQTTSN_FLAG_CLEAN;
}else{ }
else
{
_cleanSession = 0; _cleanSession = 0;
} }
} }
uint16_t LGwProxy::getNextMsgId(void){ uint16_t LGwProxy::getNextMsgId(void)
{
_nextMsgId++; _nextMsgId++;
if (_nextMsgId == 0){ if (_nextMsgId == 0)
{
_nextMsgId = 1; _nextMsgId = 1;
} }
return _nextMsgId; return _nextMsgId;
} }
void LGwProxy::checkPingReq(void){ void LGwProxy::checkPingReq(void)
{
uint8_t msg[2]; uint8_t msg[2];
msg[0] = 0x02; msg[0] = 0x02;
msg[1] = MQTTSN_TYPE_PINGREQ; msg[1] = MQTTSN_TYPE_PINGREQ;
if ( (_status == GW_CONNECTED || _status == GW_SLEPT) && isPingReqRequired() && _pingStatus != GW_WAIT_PINGRESP){ if ((_status == GW_CONNECTED || _status == GW_SLEPT) && isPingReqRequired() && _pingStatus != GW_WAIT_PINGRESP)
{
_pingStatus = GW_WAIT_PINGRESP; _pingStatus = GW_WAIT_PINGRESP;
_pingRetryCount = MQTTSN_RETRY_COUNT; _pingRetryCount = MQTTSN_RETRY_COUNT;
writeMsg((const uint8_t*) msg); writeMsg((const uint8_t*) msg);
_pingSendUTC = time(NULL); _pingSendUTC = time(NULL);
}else if (_pingStatus == GW_WAIT_PINGRESP){ }
if (_pingSendUTC + MQTTSN_TIME_RETRY < time(NULL)){ else if (_pingStatus == GW_WAIT_PINGRESP)
if (--_pingRetryCount > 0){ {
if (_pingSendUTC + MQTTSN_TIME_RETRY < time(NULL))
{
if (--_pingRetryCount > 0)
{
writeMsg((const uint8_t*) msg); writeMsg((const uint8_t*) msg);
_pingSendUTC = time(NULL); _pingSendUTC = time(NULL);
}else{ }
else
{
_status = GW_LOST; _status = GW_LOST;
_gwId = 0; _gwId = 0;
_pingStatus = 0; _pingStatus = 0;
@@ -498,8 +619,10 @@ void LGwProxy::checkPingReq(void){
} }
} }
void LGwProxy::checkAdvertise(void){ void LGwProxy::checkAdvertise(void)
if ( _gwAliveTimer.isTimeUp()){ {
if (_gwAliveTimer.isTimeUp())
{
_status = GW_LOST; _status = GW_LOST;
_gwId = 0; _gwId = 0;
_pingStatus = 0; _pingStatus = 0;
@@ -509,27 +632,37 @@ void LGwProxy::checkAdvertise(void){
} }
} }
LTopicTable* LGwProxy::getTopicTable(void){ LTopicTable* LGwProxy::getTopicTable(void)
{
return &_topicTbl; return &_topicTbl;
} }
LRegisterManager* LGwProxy::getRegisterManager(void){ LRegisterManager* LGwProxy::getRegisterManager(void)
{
return &_regMgr; return &_regMgr;
} }
bool LGwProxy::isPingReqRequired(void){ bool LGwProxy::isPingReqRequired(void)
{
return _keepAliveTimer.isTimeUp(_tkeepAlive * 1000UL); return _keepAliveTimer.isTimeUp(_tkeepAlive * 1000UL);
} }
void LGwProxy::setPingReqTimer(void){ void LGwProxy::setPingReqTimer(void)
{
_keepAliveTimer.start(_tkeepAlive * 1000UL); _keepAliveTimer.start(_tkeepAlive * 1000UL);
} }
const char* LGwProxy::getClientId(void) { const char* LGwProxy::getClientId(void)
{
return _clientId; return _clientId;
} }
void LGwProxy::setForwarderMode(void) void LGwProxy::setForwarderMode(bool valid)
{ {
_isForwarderMode = true; _isForwarderMode = valid;
}
void LGwProxy::setQoSMinus1Mode(bool valid)
{
_isQoSMinus1Mode = valid;
} }

View File

@@ -65,7 +65,8 @@ public:
void setCleanSession(bool); void setCleanSession(bool);
void setKeepAliveDuration(uint16_t duration); void setKeepAliveDuration(uint16_t duration);
void setAdvertiseDuration(uint16_t duration); void setAdvertiseDuration(uint16_t duration);
void setForwarderMode(void); void setForwarderMode(bool valid);
void setQoSMinus1Mode(bool valid);
void reconnect(void); void reconnect(void);
int writeMsg(const uint8_t* msg); int writeMsg(const uint8_t* msg);
void setPingReqTimer(void); void setPingReqTimer(void);
@@ -109,6 +110,7 @@ private:
uint16_t _tSleep; uint16_t _tSleep;
uint16_t _tWake; uint16_t _tWake;
bool _isForwarderMode; bool _isForwarderMode;
bool _isQoSMinus1Mode;
char _msg[MQTTSN_MAX_MSG_LENGTH + 1]; char _msg[MQTTSN_MAX_MSG_LENGTH + 1];
}; };

View File

@@ -27,7 +27,6 @@
======================================*/ ======================================*/
//#define DEBUG_NW //#define DEBUG_NW
//#define DEBUG_MQTTSN //#define DEBUG_MQTTSN
//#define DEBUG_OTA
/**************************************** /****************************************
MQTT-SN Parameters MQTT-SN Parameters
@@ -105,7 +104,9 @@ typedef enum
#define END_OF_SUBSCRIBE_LIST {MQTTSN_TOPIC_TYPE_NORMAL,0,0,0, 0} #define END_OF_SUBSCRIBE_LIST {MQTTSN_TOPIC_TYPE_NORMAL,0,0,0, 0}
#define UDPCONF LUdpConfig theNetcon #define UDPCONF LUdpConfig theNetcon
#define MQTTSNCONF LMqttsnConfig theMqcon #define MQTTSNCONF LMqttsnConfig theMqcon
#define SetForwarderMode theClient->getGwProxy()->setForwarderMode #define SetForwarderMode(...) theClient->getGwProxy()->setForwarderMode(__VA_ARGS__)
#define SetQoSMinus1Mode(...) theClient->getGwProxy()->setQoSMinus1Mode(__VA_ARGS__)
#ifdef CLIENT_MODE #ifdef CLIENT_MODE
#define DISPLAY(...) #define DISPLAY(...)
#define PROMPT(...) #define PROMPT(...)
@@ -142,6 +143,7 @@ typedef enum
#define QoS0 0 #define QoS0 0
#define QoS1 1 #define QoS1 1
#define QoS2 2 #define QoS2 2
#define Q0Sm1 3
#define MQTTSN_TYPE_ADVERTISE 0x00 #define MQTTSN_TYPE_ADVERTISE 0x00
#define MQTTSN_TYPE_SEARCHGW 0x01 #define MQTTSN_TYPE_SEARCHGW 0x01
#define MQTTSN_TYPE_GWINFO 0x02 #define MQTTSN_TYPE_GWINFO 0x02
@@ -177,7 +179,7 @@ typedef enum
#define MQTTSN_FLAG_QOS_0 0x0 #define MQTTSN_FLAG_QOS_0 0x0
#define MQTTSN_FLAG_QOS_1 0x20 #define MQTTSN_FLAG_QOS_1 0x20
#define MQTTSN_FLAG_QOS_2 0x40 #define MQTTSN_FLAG_QOS_2 0x40
#define MQTTSN_FLAG_QOS_N1 0xc0 #define MQTTSN_FLAG_QOS_M1 0x60
#define MQTTSN_FLAG_RETAIN 0x10 #define MQTTSN_FLAG_RETAIN 0x10
#define MQTTSN_FLAG_WILL 0x08 #define MQTTSN_FLAG_WILL 0x08
#define MQTTSN_FLAG_CLEAN 0x04 #define MQTTSN_FLAG_CLEAN 0x04

View File

@@ -75,7 +75,7 @@ void LPublishManager::publish(const char* topicName, uint8_t* payload, uint16_t
topicType = MQTTSN_TOPIC_TYPE_NORMAL; topicType = MQTTSN_TOPIC_TYPE_NORMAL;
} }
if ( qos > 0 ) if ( qos > 0 && qos < 3 )
{ {
msgId = theClient->getGwProxy()->getNextMsgId(); msgId = theClient->getGwProxy()->getNextMsgId();
} }
@@ -427,6 +427,10 @@ PubElement* LPublishManager::add(const char* topicName, uint16_t topicId, uint8_
else if (qos == 2) else if (qos == 2)
{ {
elm->flag |= MQTTSN_FLAG_QOS_2; elm->flag |= MQTTSN_FLAG_QOS_2;
}
else if (qos == 3)
{
elm->flag |= MQTTSN_FLAG_QOS_M1;
} }
if (retain) if (retain)
{ {

View File

@@ -11,6 +11,7 @@ CONFIG := gateway.conf
CLIENTS := clients.conf CLIENTS := clients.conf
PREDEFTOPIC := predefinedTopic.conf PREDEFTOPIC := predefinedTopic.conf
FORWARDERS := forwarders.conf FORWARDERS := forwarders.conf
QOSM1CLIENT := qos-1clients.conf
SRCDIR := src SRCDIR := src
SUBDIR := ../MQTTSNPacket/src SUBDIR := ../MQTTSNPacket/src
@@ -39,6 +40,7 @@ $(SRCDIR)/MQTTSNGWPublishHandler.cpp \
$(SRCDIR)/MQTTSNGWSubscribeHandler.cpp \ $(SRCDIR)/MQTTSNGWSubscribeHandler.cpp \
$(SRCDIR)/MQTTSNGWEncapsulatedPacket.cpp \ $(SRCDIR)/MQTTSNGWEncapsulatedPacket.cpp \
$(SRCDIR)/MQTTSNGWForwarder.cpp \ $(SRCDIR)/MQTTSNGWForwarder.cpp \
$(SRCDIR)/MQTTSNGWClientProxy.cpp \
$(SRCDIR)/$(OS)/$(SENSORNET)/SensorNetwork.cpp \ $(SRCDIR)/$(OS)/$(SENSORNET)/SensorNetwork.cpp \
$(SRCDIR)/$(OS)/Timer.cpp \ $(SRCDIR)/$(OS)/Timer.cpp \
$(SRCDIR)/$(OS)/Network.cpp \ $(SRCDIR)/$(OS)/Network.cpp \
@@ -141,6 +143,7 @@ install:
cp -pf $(CLIENTS) ../../ cp -pf $(CLIENTS) ../../
cp -pf $(PREDEFTOPIC) ../../ cp -pf $(PREDEFTOPIC) ../../
cp -pf $(FORWARDERS) ../../ cp -pf $(FORWARDERS) ../../
cp -pf $(QOSM1CLIENT) ../../
exectest: exectest:

View File

@@ -39,6 +39,9 @@ PredefinedTopicList=/path/to/your_predefinedTopic.conf
Forwarder=NO Forwarder=NO
ForwardersList=/home/tomoaki/tmp/forwarders.conf ForwardersList=/home/tomoaki/tmp/forwarders.conf
QoS-1=NO
QoS-1ClientsList=/path/to/your_qos-1clients.conf
#RootCAfile=/path/to/your_Root_CA.crt #RootCAfile=/path/to/your_Root_CA.crt
#RootCApath=/path/to/your_certs_directory/ #RootCApath=/path/to/your_certs_directory/
#CertKey=/path/to/your_cert.pem #CertKey=/path/to/your_cert.pem
@@ -69,6 +72,7 @@ Client should know the MulticastIP and MulticastPortNo to send a SEARCHGW messag
when **ClientAuthentication** is YES, see MQTTSNGWClient.cpp line53, clients file specified by ClientsList is required. This file defines connect allowed clients by ClientId and SensorNetwork Address. e.g. IP address and Port No. when **ClientAuthentication** is YES, see MQTTSNGWClient.cpp line53, clients file specified by ClientsList is required. This file defines connect allowed clients by ClientId and SensorNetwork Address. e.g. IP address and Port No.
When **PredefinedTopic** is YES, Pre-definedTopicID file specified by PredefinedTopicList is effective. This file defines Pre-definedTopics of the clients. In this file, ClientID,TopicName and TopicID are declared in CSV format. When **PredefinedTopic** is YES, Pre-definedTopicID file specified by PredefinedTopicList is effective. This file defines Pre-definedTopics of the clients. In this file, ClientID,TopicName and TopicID are declared in CSV format.
When **Forwarder** is YES, Forwarder Encapsulation Message is available. Connectable Forwarders are specifed by ForwardersList file. In this file, ForwarderIds and those sensorNet addresses are declared in CSV format. When **Forwarder** is YES, Forwarder Encapsulation Message is available. Connectable Forwarders are specifed by ForwardersList file. In this file, ForwarderIds and those sensorNet addresses are declared in CSV format.
When **QoS-1** is YES, QoS-1 PUBLISH Message is available. Clients which allow to send it, are specifed by QoS-1ClientsList file. In this file, ClientsId and those sensorNet addresses are declared in CSV format.

View File

@@ -11,6 +11,15 @@
# http://www.eclipse.org/org/documents/edl-v10.php. # http://www.eclipse.org/org/documents/edl-v10.php.
#*********************************************************************** #***********************************************************************
# #
# SensorNetwork address format is defined by SensorNetAddress::setAddress(string* data) function. # This file declares valid MQTTSNForwarders.
# Forwarders are defined by ForwarderId same as ClientId and those SensorNetAddress
# in a CSV format as follow:
# #
# ForwarderId, SensorNetAddress
#
# where SensorNetwork address format is defined by SensorNetAddress::setAddress(string* data) function.
#
Forwarder01,172.16.1.7:12002 Forwarder01,172.16.1.7:12002

View File

@@ -26,6 +26,9 @@ PredefinedTopicList=/path/to/your_predefinedTopic.conf
Forwarder=NO Forwarder=NO
ForwardersList=/path/to/your_forwarers.conf ForwardersList=/path/to/your_forwarers.conf
QoS-1=NO
QoS-1ClientsList=/path/to/your_qos-1clients.conf
#RootCAfile=/etc/ssl/certs/ca-certificates.crt #RootCAfile=/etc/ssl/certs/ca-certificates.crt
#RootCApath=/etc/ssl/certs/ #RootCApath=/etc/ssl/certs/
#CertsFile=/path/to/certKey.pem #CertsFile=/path/to/certKey.pem

View File

@@ -11,8 +11,27 @@
# http://www.eclipse.org/org/documents/edl-v10.php. # http://www.eclipse.org/org/documents/edl-v10.php.
#*********************************************************************** #***********************************************************************
# #
# pre-defined-topics are defined by this file.
# A format of this file is in CSV as follows:
#
# ClientID, TopicName, TopicID # ClientID, TopicName, TopicID
# #
# This file is consist from two sections.
# One for QoS-1 PUBLISH Clients, the other for another clients.
#
# pre-defined-topics for QoS-1 clients.
# ClientIDs should be "ClientProxy"
#
ClientProxy, ty4tw/proxy/predefTopic1, 1
ClientProxy, ty4tw/proxy/predefTopic2, 2
ClientProxy, ty4tw/proxy/predefTopic3, 3
#
# pre-defined-topics for another clients
#
GatewayTestClient,ty4tw/predefinedTopic1, 1 GatewayTestClient,ty4tw/predefinedTopic1, 1
GatewayTestClient,ty4tw/predefinedTopic2, 2 GatewayTestClient,ty4tw/predefinedTopic2, 2

View File

@@ -0,0 +1,27 @@
#***********************************************************************
# Copyright (c) 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
# 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.
#***********************************************************************
#
# Clients which send QoS-1 PUBLISH are defined by this file.
#
# Clients are defined by the ClientId and those SensorNetAddress
# in a CSV format as follow:
#
# ClientId, SensorNetAddress
#
# where SensorNetwork address format is defined by
# SensorNetAddress::setAddress(string* data) function.
#
#
QoS-1_Client,172.16.1.7:12002

View File

@@ -131,8 +131,7 @@ void MQTTGWPublishHandler::handlePublish(Client* client, MQTTGWPacket* packet)
/* create REGISTER */ /* create REGISTER */
MQTTSNPacket* regPacket = new MQTTSNPacket(); MQTTSNPacket* regPacket = new MQTTSNPacket();
MQTTSNString topicName; MQTTSNString topicName = MQTTSNString_initializer;
topicName.cstring = 0;
topicName.lenstring.len = topicId.data.long_.len; topicName.lenstring.len = topicId.data.long_.len;
topicName.lenstring.data = topicId.data.long_.name; topicName.lenstring.data = topicId.data.long_.name;

View File

@@ -19,12 +19,13 @@
#include "MQTTSNGWClient.h" #include "MQTTSNGWClient.h"
#include "MQTTSNGateway.h" #include "MQTTSNGateway.h"
#include "SensorNetwork.h" #include "SensorNetwork.h"
#include "MQTTSNGWForwarder.h"
#include "Network.h" #include "Network.h"
#include <string> #include <string>
#include <string.h> #include <string.h>
#include <stdio.h> #include <stdio.h>
#include "MQTTSNGWForwarder.h"
using namespace MQTTSNGW; using namespace MQTTSNGW;
char* currentDateTime(void); char* currentDateTime(void);
/*===================================== /*=====================================
@@ -80,11 +81,7 @@ bool ClientList::authorize(const char* fileName)
bool secure; bool secure;
bool stable; bool stable;
SensorNetAddress netAddr; SensorNetAddress netAddr;
MQTTSNString clientId; MQTTSNString clientId = MQTTSNString_initializer;
clientId.cstring = 0;
clientId.lenstring.data = 0;
clientId.lenstring.len = 0;
if ((fp = fopen(fileName, "r")) != 0) if ((fp = fopen(fileName, "r")) != 0)
{ {
@@ -131,13 +128,9 @@ bool ClientList::setPredefinedTopics(const char* fileName)
FILE* fp; FILE* fp;
char buf[MAX_CLIENTID_LENGTH + 256]; char buf[MAX_CLIENTID_LENGTH + 256];
size_t pos0, pos1; size_t pos0, pos1;
MQTTSNString clientId; MQTTSNString clientId = MQTTSNString_initializer;;
bool rc = false; bool rc = false;
clientId.cstring = 0;
clientId.lenstring.data = 0;
clientId.lenstring.len = 0;
if ((fp = fopen(fileName, "r")) != 0) if ((fp = fopen(fileName, "r")) != 0)
{ {
while (fgets(buf, MAX_CLIENTID_LENGTH + 254, fp) != 0) while (fgets(buf, MAX_CLIENTID_LENGTH + 254, fp) != 0)
@@ -300,9 +293,8 @@ Client* ClientList::createClient(SensorNetAddress* addr, MQTTSNString* clientId,
} }
else else
{ {
MQTTSNString dummyId; MQTTSNString dummyId MQTTSNString_initializer;;
dummyId.cstring = strdup(""); dummyId.cstring = strdup("");
dummyId.lenstring.len = 0;
client->setClientId(dummyId); client->setClientId(dummyId);
free(dummyId.cstring); free(dummyId.cstring);
} }
@@ -367,6 +359,7 @@ Client* ClientList::createPredefinedTopic( MQTTSNString* clientId, string topicN
// create Topic & Add it // create Topic & Add it
client->getTopics()->add((const char*)topicName.c_str(), topicId); client->getTopics()->add((const char*)topicName.c_str(), topicId);
client->_hasPredefTopic = true;
return client; return client;
} }
@@ -413,9 +406,11 @@ Client::Client(bool secure)
_prevClient = 0; _prevClient = 0;
_nextClient = 0; _nextClient = 0;
_clientSleepPacketQue.setMaxSize(MAX_SAVED_PUBLISH); _clientSleepPacketQue.setMaxSize(MAX_SAVED_PUBLISH);
_proxyPacketQue.setMaxSize(MAX_SAVED_PUBLISH);
_hasPredefTopic = false; _hasPredefTopic = false;
_holdPingRequest = false; _holdPingRequest = false;
_forwarder = 0; _forwarder = 0;
_isProxy = false;
} }
Client::~Client() Client::~Client()
@@ -485,6 +480,30 @@ int Client::setClientSleepPacket(MQTTGWPacket* packet)
return rc; return rc;
} }
MQTTSNPacket* Client::getProxyPacket(void)
{
return _proxyPacketQue.getPacket();
}
void Client::deleteFirstProxyPacket()
{
_proxyPacketQue.pop();
}
int Client::setProxyPacket(MQTTSNPacket* packet)
{
int rc = _proxyPacketQue.post(packet);
if ( rc )
{
WRITELOG("%s %s is Disconnected. the packet was saved.\n", currentDateTime(), _clientId);
}
else
{
WRITELOG("%s %s is Disconnected and discard the packet.\n", currentDateTime(), _clientId);
}
return rc;
}
Connect* Client::getConnectData(void) Connect* Client::getConnectData(void)
{ {
return &_connectData; return &_connectData;
@@ -572,7 +591,10 @@ void Client::updateStatus(MQTTSNPacket* packet)
case MQTTSN_PUBCOMP: case MQTTSN_PUBCOMP:
case MQTTSN_PUBREL: case MQTTSN_PUBREL:
case MQTTSN_PUBREC: case MQTTSN_PUBREC:
if ( !_isProxy )
{
_keepAliveTimer.start(_keepAliveMsec * 1.5); _keepAliveTimer.start(_keepAliveMsec * 1.5);
}
break; break;
case MQTTSN_DISCONNECT: case MQTTSN_DISCONNECT:
uint16_t duration; uint16_t duration;
@@ -641,6 +663,11 @@ void Client::disconnected(void)
_waitWillMsgFlg = false; _waitWillMsgFlg = false;
} }
void Client::tryConnect(void)
{
_status = Cstat_TryConnecting;
}
bool Client::isConnectSendable(void) bool Client::isConnectSendable(void)
{ {
if ( _status == Cstat_Lost || _status == Cstat_TryConnecting ) if ( _status == Cstat_Lost || _status == Cstat_TryConnecting )
@@ -703,6 +730,11 @@ void Client::setTopics(Topics* topics)
_topics = topics; _topics = topics;
} }
ClientStatus Client::getClientStatus(void)
{
return _status;
}
void Client::setWaitWillMsgFlg(bool flg) void Client::setWaitWillMsgFlg(bool flg)
{ {
_waitWillMsgFlg = flg; _waitWillMsgFlg = flg;
@@ -733,6 +765,11 @@ bool Client::isAwake(void)
return (_status == Cstat_Awake); return (_status == Cstat_Awake);
} }
bool Client::isConnecting(void)
{
return (_status == Cstat_Connecting);
}
bool Client::isSecureNetwork(void) bool Client::isSecureNetwork(void)
{ {
return _secureNetwork; return _secureNetwork;
@@ -760,11 +797,19 @@ void Client::setClientId(MQTTSNString id)
free(_clientId); free(_clientId);
} }
if ( id.cstring )
{
_clientId = (char*)calloc(strlen(id.cstring) + 1, 1);
memcpy(_clientId, id.cstring, strlen(id.cstring));
}
else
{
/* save clientId into (char*)_clientId NULL terminated */ /* save clientId into (char*)_clientId NULL terminated */
_clientId = (char*)calloc(MQTTSNstrlen(id) + 1, 1); _clientId = (char*)calloc(MQTTSNstrlen(id) + 1, 1);
unsigned char* ptr = (unsigned char*)_clientId; unsigned char* ptr = (unsigned char*)_clientId;
writeMQTTSNString((unsigned char**)&ptr, id); writeMQTTSNString((unsigned char**)&ptr, id);
} }
}
void Client::setWillTopic(MQTTSNString willTopic) void Client::setWillTopic(MQTTSNString willTopic)
{ {
@@ -812,14 +857,14 @@ const char* Client::getStatus(void)
return theClientStatus[_status]; return theClientStatus[_status];
} }
Client* Client::getOTAClient(void) bool Client::isProxy(void)
{ {
return _otaClient; return _isProxy;
} }
void Client::setOTAClient(Client* cl) void Client::setPorxy(bool isProxy)
{ {
_otaClient =cl; _isProxy = isProxy;;
} }
void Client::holdPingRequest(void) void Client::holdPingRequest(void)

View File

@@ -253,6 +253,9 @@ public:
TopicIdMapelement* getWaitedSubTopicId(uint16_t msgId); TopicIdMapelement* getWaitedSubTopicId(uint16_t msgId);
MQTTGWPacket* getClientSleepPacket(void); MQTTGWPacket* getClientSleepPacket(void);
void deleteFirstClientSleepPacket(void); void deleteFirstClientSleepPacket(void);
MQTTSNPacket* getProxyPacket(void);
void deleteFirstProxyPacket(void);
WaitREGACKPacketList* getWaitREGACKPacketList(void); WaitREGACKPacketList* getWaitREGACKPacketList(void);
void eraseWaitedPubTopicId(uint16_t msgId); void eraseWaitedPubTopicId(uint16_t msgId);
@@ -261,6 +264,7 @@ public:
void clearWaitedSubTopicId(void); void clearWaitedSubTopicId(void);
int setClientSleepPacket(MQTTGWPacket*); int setClientSleepPacket(MQTTGWPacket*);
int setProxyPacket(MQTTSNPacket* packet);
void setWaitedPubTopicId(uint16_t msgId, uint16_t topicId, MQTTSN_topicTypes type); void setWaitedPubTopicId(uint16_t msgId, uint16_t topicId, MQTTSN_topicTypes type);
void setWaitedSubTopicId(uint16_t msgId, uint16_t topicId, MQTTSN_topicTypes type); void setWaitedSubTopicId(uint16_t msgId, uint16_t topicId, MQTTSN_topicTypes type);
@@ -271,6 +275,8 @@ public:
void connackSended(int rc); void connackSended(int rc);
void disconnected(void); void disconnected(void);
bool isConnectSendable(void); bool isConnectSendable(void);
void tryConnect(void);
ClientStatus getClientStatus(void);
uint16_t getNextPacketId(void); uint16_t getNextPacketId(void);
uint8_t getNextSnMsgId(void); uint8_t getNextSnMsgId(void);
@@ -286,6 +292,9 @@ public:
Forwarder* getForwarder(void); Forwarder* getForwarder(void);
void setForwarder(Forwarder* forwader); void setForwarder(Forwarder* forwader);
void setPorxy(bool isProxy);
bool isProxy(void);
void setClientId(MQTTSNString id); void setClientId(MQTTSNString id);
void setWillTopic(MQTTSNString willTopic); void setWillTopic(MQTTSNString willTopic);
void setWillMsg(MQTTSNString willmsg); void setWillMsg(MQTTSNString willmsg);
@@ -298,6 +307,7 @@ public:
bool erasable(void); bool erasable(void);
bool isDisconnect(void); bool isDisconnect(void);
bool isConnecting(void);
bool isActive(void); bool isActive(void);
bool isSleep(void); bool isSleep(void);
bool isAwake(void); bool isAwake(void);
@@ -310,11 +320,11 @@ public:
bool isHoldPringReqest(void); bool isHoldPringReqest(void);
Client* getNextClient(void); Client* getNextClient(void);
Client* getOTAClient(void);
void setOTAClient(Client* cl);
private: private:
PacketQue<MQTTGWPacket> _clientSleepPacketQue; PacketQue<MQTTGWPacket> _clientSleepPacketQue;
PacketQue<MQTTSNPacket> _proxyPacketQue;
WaitREGACKPacketList _waitREGACKList; WaitREGACKPacketList _waitREGACKList;
Topics* _topics; Topics* _topics;
@@ -345,6 +355,7 @@ private:
SensorNetAddress _sensorNetAddr; SensorNetAddress _sensorNetAddr;
Forwarder* _forwarder; Forwarder* _forwarder;
bool _isProxy;
bool _sessionStatus; bool _sessionStatus;

View File

@@ -0,0 +1,252 @@
/**************************************************************************************
* Copyright (c) 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
* 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:
* Tomoaki Yamaguchi - initial API and implementation and/or initial documentation
**************************************************************************************/
#include "MQTTSNGWDefines.h"
#include "MQTTSNGWClientProxy.h"
#include "MQTTSNGateway.h"
#include "SensorNetwork.h"
#include <string>
#include <string.h>
#include <stdio.h>
using namespace MQTTSNGW;
#define RESPONSE_DURATION 900 // Secs
/*
* Class ClientProxyElement
*/
ClientProxyElement::ClientProxyElement(void)
: _clientId{0}
, _next{0}
{
}
ClientProxyElement::ClientProxyElement(SensorNetAddress* addr, string* clientId)
: _next{0}
{
_clientId = *clientId;
_sensorNetAddr = *addr;
}
ClientProxyElement::~ClientProxyElement(void)
{
}
/*
* Class ClientProxy
*/
ClientProxy:: ClientProxy(void)
: _head{0}
{
_gateway = 0;
_client = 0;
}
ClientProxy:: ClientProxy(Gateway* gw)
: _head{0}
{
_gateway = gw;
_client = 0;
}
ClientProxy::~ClientProxy(void)
{
if ( _head )
{
ClientProxyElement* p = _head;
while ( p )
{
ClientProxyElement* next = p->_next;
delete p;
p = next;
}
}
}
void ClientProxy::setGateway(Gateway* gw)
{
_gateway = gw;
}
ClientProxyElement* ClientProxy::add(SensorNetAddress* addr, string* clientId)
{
ClientProxyElement* elm = new ClientProxyElement(addr, clientId);
if ( _head == 0 )
{
_head = elm;
}
else
{
ClientProxyElement* p = _head;
while ( p )
{
if ( p->_next == 0 )
{
p->_next = elm;
break;
}
else
{
p = p->_next;
}
}
}
return elm;
}
const char* ClientProxy::getClientId(SensorNetAddress* addr)
{
ClientProxyElement* p = _head;
while ( p )
{
if ( p->_sensorNetAddr.isMatch(addr) )
{
return p->_clientId.c_str();
break;
}
p = p->_next;
}
return 0;
}
void ClientProxy::setClient(Client* client)
{
_client = client;
}
Client* ClientProxy::getClient(void)
{
return _client;
}
bool ClientProxy::setClientProxy(const char* fileName)
{
FILE* fp;
char buf[MAX_CLIENTID_LENGTH + 256];
size_t pos;
SensorNetAddress netAddr;
if ((fp = fopen(fileName, "r")) != 0)
{
while (fgets(buf, MAX_CLIENTID_LENGTH + 254, fp) != 0)
{
if (*buf == '#')
{
continue;
}
string data = string(buf);
while ((pos = data.find_first_of("  \t\n")) != string::npos)
{
data.erase(pos, 1);
}
if (data.empty())
{
continue;
}
pos = data.find_first_of(",");
string id = data.substr(0, pos);
string addr = data.substr(pos + 1);
if (netAddr.setAddress(&addr) == 0)
{
add(&netAddr, &id);
}
else
{
WRITELOG("Invalid address %s\n", data.c_str());
return false;
}
}
fclose(fp);
}
else
{
WRITELOG("Can not open the QoS_1Client List. %s\n", fileName);
return false;
}
return true;
}
void ClientProxy::checkConnection(void)
{
if ( _client->isDisconnect() || ( _client->isConnecting() && _responseTimer.isTimeup()) )
{
_client->connectSended();
_responseTimer.start(RESPONSE_DURATION * 1000UL);
MQTTSNPacket_connectData options = MQTTSNPacket_connectData_initializer;
options.clientID.cstring = _client->getClientId();
options.duration = RESPONSE_DURATION;
MQTTSNPacket* packet = new MQTTSNPacket();
packet->setCONNECT(&options);
Event* ev = new Event();
ev->setClientRecvEvent(_client, packet);
_gateway->getPacketEventQue()->post(ev);
}
else if ( _client->isActive() && _keepAliveTimer.isTimeup() )
{
MQTTSNPacket* packet = new MQTTSNPacket();
MQTTSNString clientId = MQTTSNString_initializer;
packet->setPINGREQ(&clientId);
Event* ev = new Event();
ev->setClientRecvEvent(_client, packet);
_gateway->getPacketEventQue()->post(ev);
resetPingTimer();
}
}
void ClientProxy::resetPingTimer(void)
{
_keepAliveTimer.start(RESPONSE_DURATION * 1000UL);
}
void ClientProxy::send(MQTTSNPacket* packet)
{
if ( packet->getType() == MQTTSN_CONNACK || packet->getType() == MQTTSN_PINGRESP )
{
resetPingTimer();
sendStoredPublish();
}
else if ( packet->getType() == MQTTSN_PINGRESP )
{
resetPingTimer();
}
}
void ClientProxy::sendStoredPublish(void)
{
MQTTSNPacket* msg = 0;
while ( ( msg = _client->getProxyPacket() ) != 0 )
{
_client->deleteFirstProxyPacket(); // pop the que to delete element.
Event* ev = new Event();
ev->setClientRecvEvent(_client, msg);
_gateway->getPacketEventQue()->post(ev);
}
}

View File

@@ -0,0 +1,77 @@
/**************************************************************************************
* Copyright (c) 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
* 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:
* Tomoaki Yamaguchi - initial API and implementation and/or initial documentation
**************************************************************************************/
#ifndef MQTTSNGATEWAY_SRC_MQTTSNGWCLIENTPROXY_H_
#define MQTTSNGATEWAY_SRC_MQTTSNGWCLIENTPROXY_H_
#include "MQTTSNGateway.h"
#include "MQTTGWPacket.h"
#include "MQTTSNGWClient.h"
#include "SensorNetwork.h"
#include "MQTTSNGWProcess.h"
namespace MQTTSNGW
{
class Gateway;
class ClientProxyElement
{
friend class ClientProxy;
public:
ClientProxyElement(void);
ClientProxyElement(SensorNetAddress* addr, string* clientId);
~ClientProxyElement(void);
private:
SensorNetAddress _sensorNetAddr;
string _clientId;
ClientProxyElement* _next;
};
class ClientProxy
{
public:
ClientProxy(void);
ClientProxy(Gateway* gw);
~ClientProxy(void);
bool setClientProxy(const char* fileName);
ClientProxyElement* add(SensorNetAddress* addr, string* clientId);
const char* getClientId(SensorNetAddress* addr);
void setClient(Client*);
Client* getClient(void);
void setGateway(Gateway* gw);
void setKeepAlive(uint16_t secs);
void checkConnection(void);
void resetPingTimer(void);
void send(MQTTSNPacket* packet);
private:
void sendStoredPublish(void);
Gateway* _gateway;
Client* _client;
ClientProxyElement* _head;
Timer _keepAliveTimer;
Timer _responseTimer;
};
}
#endif /* MQTTSNGATEWAY_SRC_MQTTSNGWCLIENTPROXY_H_ */

View File

@@ -17,10 +17,11 @@
#include "MQTTSNGWClientRecvTask.h" #include "MQTTSNGWClientRecvTask.h"
#include "MQTTSNGateway.h" #include "MQTTSNGateway.h"
#include "MQTTSNPacket.h" #include "MQTTSNPacket.h"
#include "MQTTSNGWForwarder.h"
#include "MQTTSNGWEncapsulatedPacket.h" #include "MQTTSNGWEncapsulatedPacket.h"
#include <cstring> #include <cstring>
#include "MQTTSNGWForwarder.h"
using namespace MQTTSNGW; using namespace MQTTSNGW;
char* currentDateTime(void); char* currentDateTime(void);
/*===================================== /*=====================================
@@ -91,31 +92,32 @@ void ClientRecvTask::run()
if ( packet->getType() == MQTTSN_SEARCHGW ) if ( packet->getType() == MQTTSN_SEARCHGW )
{ {
/* write log and post Event */ /* write log and post Event */
log(0, packet); log(0, packet, 0);
ev = new Event(); ev = new Event();
ev->setBrodcastEvent(packet); ev->setBrodcastEvent(packet);
_gateway->getPacketEventQue()->post(ev); _gateway->getPacketEventQue()->post(ev);
continue; continue;
} }
if ( packet->getType() == MQTTSN_ENCAPSULATED ) if ( packet->getType() == MQTTSN_ENCAPSULATED )
{ {
fwd = _gateway->getForwarderList()->getForwarder(_sensorNetwork->getSenderAddress()); fwd = _gateway->getForwarderList()->getForwarder(_sensorNetwork->getSenderAddress());
if ( fwd == 0 ) if ( fwd == 0 )
{ {
log(0, packet); log(0, packet, 0);
WRITELOG("%s Forwarder %s is not authorized.%s\n", ERRMSG_HEADER, _sensorNetwork->getSenderAddress()->sprint(buf), ERRMSG_FOOTER); WRITELOG("%s Forwarder %s is not authenticated.%s\n", ERRMSG_HEADER, _sensorNetwork->getSenderAddress()->sprint(buf), ERRMSG_FOOTER);
delete packet; delete packet;
continue; continue;
} }
else else
{ {
MQTTSNString fwdName; MQTTSNString fwdName = MQTTSNString_initializer;
fwdName.lenstring.data = const_cast<char *>( fwd->getName() ); fwdName.cstring = const_cast<char *>( fwd->getName() );
fwdName.lenstring.len = strlen(fwdName.lenstring.data);
log(0, packet, &fwdName); log(0, packet, &fwdName);
/* get the packet from the encapsulation message */
MQTTSNGWEncapsulatedPacket encap; MQTTSNGWEncapsulatedPacket encap;
encap.desirialize(packet->getPacketData(), packet->getPacketLength()); encap.desirialize(packet->getPacketData(), packet->getPacketLength());
nodeId.setId( encap.getWirelessNodeId() ); nodeId.setId( encap.getWirelessNodeId() );
@@ -125,16 +127,36 @@ void ClientRecvTask::run()
} }
} }
else else
{
const char* clientName = _gateway->getClientProxy()->getClientId(_sensorNetwork->getSenderAddress());
if ( clientName ) // This client is for QoS-1 PUBLISH.
{
if ( packet->isQoSMinusPUBLISH() )
{
client = _gateway->getClientProxy()->getClient(); // point to the ClientProxy
}
else
{
client = _gateway->getClientProxy()->getClient();
log(clientName, packet);
WRITELOG("%s %s %s can send only PUBLISH with QoS-1.%s\n", ERRMSG_HEADER, clientName, _sensorNetwork->getSenderAddress()->sprint(buf), ERRMSG_FOOTER);
delete packet;
continue;
}
}
else
{ {
/* get client from the ClientList of Gateway by sensorNetAddress. */ /* get client from the ClientList of Gateway by sensorNetAddress. */
client = _gateway->getClientList()->getClient(_sensorNetwork->getSenderAddress()); client = _gateway->getClientList()->getClient(_sensorNetwork->getSenderAddress());
} }
}
if ( client ) if ( client )
{ {
/* write log and post Event */ /* write log and post Event */
log(client, packet); log(client, packet, 0);
ev = new Event(); ev = new Event();
ev->setClientRecvEvent(client,packet); ev->setClientRecvEvent(client,packet);
_gateway->getPacketEventQue()->post(ev); _gateway->getPacketEventQue()->post(ev);
@@ -196,7 +218,7 @@ void ClientRecvTask::run()
} }
else else
{ {
log(client, packet); log(client, packet, 0);
WRITELOG("%s Client(%s) is not connecting. message has been discarded.%s\n", ERRMSG_HEADER, _sensorNetwork->getSenderAddress()->sprint(buf), ERRMSG_FOOTER); WRITELOG("%s Client(%s) is not connecting. message has been discarded.%s\n", ERRMSG_HEADER, _sensorNetwork->getSenderAddress()->sprint(buf), ERRMSG_FOOTER);
delete packet; delete packet;
@@ -216,17 +238,23 @@ void ClientRecvTask::run()
void ClientRecvTask::log(Client* client, MQTTSNPacket* packet, MQTTSNString* id) void ClientRecvTask::log(Client* client, MQTTSNPacket* packet, MQTTSNString* id)
{ {
char pbuf[SIZE_OF_LOG_PACKET * 3];
const char* clientId; const char* clientId;
char cstr[MAX_CLIENTID_LENGTH + 1]; char cstr[MAX_CLIENTID_LENGTH + 1];
char msgId[6];
if ( id ) if ( id )
{
if ( id->cstring )
{
strncpy(cstr, id->cstring, strlen(id->cstring) );
clientId = cstr;
}
else
{ {
memset((void*)cstr, 0, id->lenstring.len + 1); memset((void*)cstr, 0, id->lenstring.len + 1);
strncpy(cstr, id->lenstring.data, id->lenstring.len ); strncpy(cstr, id->lenstring.data, id->lenstring.len );
clientId = cstr; clientId = cstr;
} }
}
else if ( client ) else if ( client )
{ {
clientId = client->getClientId(); clientId = client->getClientId();
@@ -236,6 +264,14 @@ void ClientRecvTask::log(Client* client, MQTTSNPacket* packet, MQTTSNString* id)
clientId = UNKNOWNCL; clientId = UNKNOWNCL;
} }
log(clientId, packet);
}
void ClientRecvTask::log(const char* clientId, MQTTSNPacket* packet)
{
char pbuf[SIZE_OF_LOG_PACKET * 3];
char msgId[6];
switch (packet->getType()) switch (packet->getType())
{ {
case MQTTSN_SEARCHGW: case MQTTSN_SEARCHGW:

View File

@@ -35,8 +35,8 @@ public:
void run(); void run();
private: private:
void log(Client*, MQTTSNPacket*, MQTTSNString* id = 0); void log(Client*, MQTTSNPacket*, MQTTSNString* id);
void log(const char* clientId, MQTTSNPacket* packet);
Gateway* _gateway; Gateway* _gateway;
SensorNetwork* _sensorNetwork; SensorNetwork* _sensorNetwork;
}; };

View File

@@ -71,6 +71,11 @@ void ClientSendTask::run()
else else
{ {
log(client, packet); log(client, packet);
if ( client->isProxy() )
{
_gateway->getClientProxy()->send(packet);
continue;
}
rc = packet->unicast(_sensorNetwork, client->getSensorNetAddress()); rc = packet->unicast(_sensorNetwork, client->getSensorNetAddress());
} }
} }

View File

@@ -90,7 +90,10 @@ void MQTTSNConnectionHandler::handleConnect(Client* client, MQTTSNPacket* packet
//* clear ConnectData of Client */ //* clear ConnectData of Client */
Connect* connectData = client->getConnectData(); Connect* connectData = client->getConnectData();
memset(connectData, 0, sizeof(Connect)); memset(connectData, 0, sizeof(Connect));
if ( !client->isProxy() )
{
client->disconnected(); client->disconnected();
}
Topics* topics = client->getTopics(); Topics* topics = client->getTopics();
@@ -153,7 +156,7 @@ void MQTTSNConnectionHandler::handleWilltopic(Client* client, MQTTSNPacket* pack
{ {
int willQos; int willQos;
uint8_t willRetain; uint8_t willRetain;
MQTTSNString willTopic; MQTTSNString willTopic = MQTTSNString_initializer;
if ( packet->getWILLTOPIC(&willQos, &willRetain, &willTopic) == 0 ) if ( packet->getWILLTOPIC(&willQos, &willRetain, &willTopic) == 0 )
{ {
@@ -187,7 +190,7 @@ void MQTTSNConnectionHandler::handleWillmsg(Client* client, MQTTSNPacket* packet
return; return;
} }
MQTTSNString willmsg; MQTTSNString willmsg = MQTTSNString_initializer;
Connect* connectData = client->getConnectData(); Connect* connectData = client->getConnectData();
if( client->isConnectSendable() ) if( client->isConnectSendable() )

View File

@@ -27,6 +27,7 @@ namespace MQTTSNGW
#define CLIENT_LIST "clients.conf" #define CLIENT_LIST "clients.conf"
#define PREDEFINEDTOPIC_FILE "predefinedTopic.conf" #define PREDEFINEDTOPIC_FILE "predefinedTopic.conf"
#define FORWARDER_LIST "forwarders.conf" #define FORWARDER_LIST "forwarders.conf"
#define QOS_1CLIENT_LIST "qos-1clients.conf"
/*========================================================== /*==========================================================
* Gateway default parameters * Gateway default parameters

View File

@@ -13,9 +13,10 @@
* Contributors: * Contributors:
* Tomoaki Yamaguchi - initial API and implementation and/or initial documentation * Tomoaki Yamaguchi - initial API and implementation and/or initial documentation
**************************************************************************************/ **************************************************************************************/
#include <string.h>
#include "MQTTSNGWForwarder.h" #include "MQTTSNGWForwarder.h"
#include <string.h>
using namespace MQTTSNGW; using namespace MQTTSNGW;
using namespace std; using namespace std;
@@ -155,10 +156,10 @@ Forwarder::~Forwarder(void)
{ {
if ( _headClient ) if ( _headClient )
{ {
ForwardedClient* p = _headClient; ForwarderElement* p = _headClient;
while ( p ) while ( p )
{ {
ForwardedClient* next = p->_next; ForwarderElement* next = p->_next;
delete p; delete p;
p = next; p = next;
} }
@@ -172,8 +173,8 @@ const char* Forwarder::getId(void)
void Forwarder::addClient(Client* client, WirelessNodeId* id) void Forwarder::addClient(Client* client, WirelessNodeId* id)
{ {
ForwardedClient* p = _headClient; ForwarderElement* p = _headClient;
ForwardedClient* prev = 0; ForwarderElement* prev = 0;
client->setForwarder(this); client->setForwarder(this);
@@ -191,7 +192,7 @@ void Forwarder::addClient(Client* client, WirelessNodeId* id)
} }
} }
ForwardedClient* fclient = new ForwardedClient(); ForwarderElement* fclient = new ForwarderElement();
fclient->setClient(client); fclient->setClient(client);
fclient->setWirelessNodeId(id); fclient->setWirelessNodeId(id);
@@ -210,7 +211,7 @@ Client* Forwarder::getClient(WirelessNodeId* id)
{ {
Client* cl = 0; Client* cl = 0;
_mutex.lock(); _mutex.lock();
ForwardedClient* p = _headClient; ForwarderElement* p = _headClient;
while ( p ) while ( p )
{ {
if ( *(p->_wirelessNodeId) == *id ) if ( *(p->_wirelessNodeId) == *id )
@@ -236,7 +237,7 @@ WirelessNodeId* Forwarder::getWirelessNodeId(Client* client)
{ {
WirelessNodeId* nodeId = 0; WirelessNodeId* nodeId = 0;
_mutex.lock(); _mutex.lock();
ForwardedClient* p = _headClient; ForwarderElement* p = _headClient;
while ( p ) while ( p )
{ {
if ( p->_client == client ) if ( p->_client == client )
@@ -255,9 +256,9 @@ WirelessNodeId* Forwarder::getWirelessNodeId(Client* client)
void Forwarder::eraseClient(Client* client) void Forwarder::eraseClient(Client* client)
{ {
ForwardedClient* prev = 0; ForwarderElement* prev = 0;
_mutex.lock(); _mutex.lock();
ForwardedClient* p = _headClient; ForwarderElement* p = _headClient;
while ( p ) while ( p )
{ {
@@ -291,14 +292,14 @@ SensorNetAddress* Forwarder::getSensorNetAddr(void)
* Class ForwardedClient * Class ForwardedClient
*/ */
ForwardedClient::ForwardedClient() ForwarderElement::ForwarderElement()
: _client{0} : _client{0}
, _wirelessNodeId{0} , _wirelessNodeId{0}
, _next{0} , _next{0}
{ {
} }
ForwardedClient::~ForwardedClient() ForwarderElement::~ForwarderElement()
{ {
if (_wirelessNodeId) if (_wirelessNodeId)
{ {
@@ -306,12 +307,12 @@ ForwardedClient::~ForwardedClient()
} }
} }
void ForwardedClient::setClient(Client* client) void ForwarderElement::setClient(Client* client)
{ {
_client = client; _client = client;
} }
void ForwardedClient::setWirelessNodeId(WirelessNodeId* id) void ForwarderElement::setWirelessNodeId(WirelessNodeId* id)
{ {
if ( _wirelessNodeId == 0 ) if ( _wirelessNodeId == 0 )
{ {

View File

@@ -28,18 +28,18 @@ namespace MQTTSNGW
class Client; class Client;
class WirelessNodeId; class WirelessNodeId;
class ForwardedClient class ForwarderElement
{ {
friend class Forwarder; friend class Forwarder;
public: public:
ForwardedClient(); ForwarderElement();
~ForwardedClient(); ~ForwarderElement();
void setClient(Client* client); void setClient(Client* client);
void setWirelessNodeId(WirelessNodeId* id); void setWirelessNodeId(WirelessNodeId* id);
private: private:
Client* _client; Client* _client;
WirelessNodeId* _wirelessNodeId; WirelessNodeId* _wirelessNodeId;
ForwardedClient* _next; ForwarderElement* _next;
}; };
@@ -62,7 +62,7 @@ public:
private: private:
string _forwarderName; string _forwarderName;
SensorNetAddress _sensorNetAddr; SensorNetAddress _sensorNetAddr;
ForwardedClient* _headClient; ForwarderElement* _headClient;
Forwarder* _next; Forwarder* _next;
Mutex _mutex; Mutex _mutex;
}; };

View File

@@ -101,6 +101,17 @@ int MQTTSNPacket::getType(void)
return _buf[p]; return _buf[p];
} }
bool MQTTSNPacket::isQoSMinusPUBLISH(void)
{
if ( _bufLen == 0 )
{
return false;;
}
int value = 0;
int p = MQTTSNPacket_decode(_buf, _bufLen, &value);
return ( (_buf[p] == MQTTSN_PUBLISH) && ((_buf[p + 1] & 0x60 ) == 0x60 ));
}
unsigned char* MQTTSNPacket::getPacketData(void) unsigned char* MQTTSNPacket::getPacketData(void)
{ {
return _buf; return _buf;
@@ -119,7 +130,8 @@ const char* MQTTSNPacket::getName()
int MQTTSNPacket::setADVERTISE(uint8_t gatewayid, uint16_t duration) int MQTTSNPacket::setADVERTISE(uint8_t gatewayid, uint16_t duration)
{ {
unsigned char buf[5]; unsigned char buf[5];
int len = MQTTSNSerialize_advertise(buf, 5, (unsigned char) gatewayid, int buflen = sizeof(buf);
int len = MQTTSNSerialize_advertise(buf, buflen, (unsigned char) gatewayid,
(unsigned short) duration); (unsigned short) duration);
return desirialize(buf, len); return desirialize(buf, len);
} }
@@ -127,44 +139,50 @@ int MQTTSNPacket::setADVERTISE(uint8_t gatewayid, uint16_t duration)
int MQTTSNPacket::setGWINFO(uint8_t gatewayId) int MQTTSNPacket::setGWINFO(uint8_t gatewayId)
{ {
unsigned char buf[3]; unsigned char buf[3];
int len = MQTTSNSerialize_gwinfo(buf, 3, (unsigned char) gatewayId, 0, 0); int buflen = sizeof(buf);
int len = MQTTSNSerialize_gwinfo(buf, buflen, (unsigned char) gatewayId, 0, 0);
return desirialize(buf, len); return desirialize(buf, len);
} }
int MQTTSNPacket::setConnect(void) int MQTTSNPacket::setConnect(void)
{ {
unsigned char buf[40]; unsigned char buf[40];
int buflen = sizeof(buf);
MQTTSNPacket_connectData data; MQTTSNPacket_connectData data;
data.clientID.cstring = (char*)"client01"; data.clientID.cstring = (char*)"client01";
int len = MQTTSNSerialize_connect(buf, 40, &data); int len = MQTTSNSerialize_connect(buf, buflen, &data);
return desirialize(buf, len); return desirialize(buf, len);
} }
int MQTTSNPacket::setCONNACK(uint8_t returnCode) int MQTTSNPacket::setCONNACK(uint8_t returnCode)
{ {
unsigned char buf[3]; unsigned char buf[3];
int len = MQTTSNSerialize_connack(buf, 3, (int) returnCode); int buflen = sizeof(buf);
int len = MQTTSNSerialize_connack(buf, buflen, (int) returnCode);
return desirialize(buf, len); return desirialize(buf, len);
} }
int MQTTSNPacket::setWILLTOPICREQ(void) int MQTTSNPacket::setWILLTOPICREQ(void)
{ {
unsigned char buf[2]; unsigned char buf[2];
int len = MQTTSNSerialize_willtopicreq(buf, 2); int buflen = sizeof(buf);
int len = MQTTSNSerialize_willtopicreq(buf, buflen);
return desirialize(buf, len); return desirialize(buf, len);
} }
int MQTTSNPacket::setWILLMSGREQ(void) int MQTTSNPacket::setWILLMSGREQ(void)
{ {
unsigned char buf[2]; unsigned char buf[2];
int len = MQTTSNSerialize_willmsgreq(buf, 2); int buflen = sizeof(buf);
int len = MQTTSNSerialize_willmsgreq(buf, buflen);
return desirialize(buf, len); return desirialize(buf, len);
} }
int MQTTSNPacket::setREGISTER(uint16_t topicId, uint16_t msgId, MQTTSNString* topicName) int MQTTSNPacket::setREGISTER(uint16_t topicId, uint16_t msgId, MQTTSNString* topicName)
{ {
unsigned char buf[MQTTSNGW_MAX_PACKET_SIZE]; unsigned char buf[MQTTSNGW_MAX_PACKET_SIZE];
int len = MQTTSNSerialize_register(buf, MQTTSNGW_MAX_PACKET_SIZE, (unsigned short) topicId, (unsigned short) msgId, int buflen = sizeof(buf);
int len = MQTTSNSerialize_register(buf, buflen, (unsigned short) topicId, (unsigned short) msgId,
topicName); topicName);
return desirialize(buf, len); return desirialize(buf, len);
} }
@@ -172,7 +190,8 @@ int MQTTSNPacket::setREGISTER(uint16_t topicId, uint16_t msgId, MQTTSNString* to
int MQTTSNPacket::setREGACK(uint16_t topicId, uint16_t msgId, uint8_t returnCode) int MQTTSNPacket::setREGACK(uint16_t topicId, uint16_t msgId, uint8_t returnCode)
{ {
unsigned char buf[7]; unsigned char buf[7];
int len = MQTTSNSerialize_regack(buf, 7, (unsigned short) topicId, (unsigned short) msgId, int buflen = sizeof(buf);
int len = MQTTSNSerialize_regack(buf, buflen, (unsigned short) topicId, (unsigned short) msgId,
(unsigned char) returnCode); (unsigned char) returnCode);
return desirialize(buf, len); return desirialize(buf, len);
} }
@@ -181,7 +200,8 @@ int MQTTSNPacket::setPUBLISH(uint8_t dup, int qos, uint8_t retained, uint16_t ms
uint8_t* payload, uint16_t payloadlen) uint8_t* payload, uint16_t payloadlen)
{ {
unsigned char buf[MQTTSNGW_MAX_PACKET_SIZE]; unsigned char buf[MQTTSNGW_MAX_PACKET_SIZE];
int len = MQTTSNSerialize_publish(buf, MQTTSNGW_MAX_PACKET_SIZE, (unsigned char) dup, qos, (unsigned char) retained, int buflen = sizeof(buf);
int len = MQTTSNSerialize_publish(buf, buflen, (unsigned char) dup, qos, (unsigned char) retained,
(unsigned short) msgId, topic, (unsigned char*) payload, (int) payloadlen); (unsigned short) msgId, topic, (unsigned char*) payload, (int) payloadlen);
return desirialize(buf, len); return desirialize(buf, len);
} }
@@ -189,7 +209,8 @@ int MQTTSNPacket::setPUBLISH(uint8_t dup, int qos, uint8_t retained, uint16_t ms
int MQTTSNPacket::setPUBACK(uint16_t topicId, uint16_t msgId, uint8_t returnCode) int MQTTSNPacket::setPUBACK(uint16_t topicId, uint16_t msgId, uint8_t returnCode)
{ {
unsigned char buf[7]; unsigned char buf[7];
int len = MQTTSNSerialize_puback(buf, 7, (unsigned short) topicId, (unsigned short) msgId, int buflen = sizeof(buf);
int len = MQTTSNSerialize_puback(buf, buflen, (unsigned short) topicId, (unsigned short) msgId,
(unsigned char) returnCode); (unsigned char) returnCode);
return desirialize(buf, len); return desirialize(buf, len);
} }
@@ -197,28 +218,32 @@ int MQTTSNPacket::setPUBACK(uint16_t topicId, uint16_t msgId, uint8_t returnCode
int MQTTSNPacket::setPUBREC(uint16_t msgId) int MQTTSNPacket::setPUBREC(uint16_t msgId)
{ {
unsigned char buf[4]; unsigned char buf[4];
int len = MQTTSNSerialize_pubrec(buf, 4, (unsigned short) msgId); int buflen = sizeof(buf);
int len = MQTTSNSerialize_pubrec(buf, buflen, (unsigned short) msgId);
return desirialize(buf, len); return desirialize(buf, len);
} }
int MQTTSNPacket::setPUBREL(uint16_t msgId) int MQTTSNPacket::setPUBREL(uint16_t msgId)
{ {
unsigned char buf[4]; unsigned char buf[4];
int len = MQTTSNSerialize_pubrel(buf, 4, (unsigned short) msgId); int buflen = sizeof(buf);
int len = MQTTSNSerialize_pubrel(buf, buflen, (unsigned short) msgId);
return desirialize(buf, len); return desirialize(buf, len);
} }
int MQTTSNPacket::setPUBCOMP(uint16_t msgId) int MQTTSNPacket::setPUBCOMP(uint16_t msgId)
{ {
unsigned char buf[4]; unsigned char buf[4];
int len = MQTTSNSerialize_pubcomp(buf, 4, (unsigned short) msgId); int buflen = sizeof(buf);
int len = MQTTSNSerialize_pubcomp(buf, buflen, (unsigned short) msgId);
return desirialize(buf, len); return desirialize(buf, len);
} }
int MQTTSNPacket::setSUBACK(int qos, uint16_t topicId, uint16_t msgId, uint8_t returnCode) int MQTTSNPacket::setSUBACK(int qos, uint16_t topicId, uint16_t msgId, uint8_t returnCode)
{ {
unsigned char buf[8]; unsigned char buf[8];
int len = MQTTSNSerialize_suback(buf, 8, qos, (unsigned short) topicId, int buflen = sizeof(buf);
int len = MQTTSNSerialize_suback(buf, buflen, qos, (unsigned short) topicId,
(unsigned short) msgId, (unsigned char) returnCode); (unsigned short) msgId, (unsigned char) returnCode);
return desirialize(buf, len); return desirialize(buf, len);
} }
@@ -226,35 +251,56 @@ int MQTTSNPacket::setSUBACK(int qos, uint16_t topicId, uint16_t msgId, uint8_t r
int MQTTSNPacket::setUNSUBACK(uint16_t msgId) int MQTTSNPacket::setUNSUBACK(uint16_t msgId)
{ {
unsigned char buf[4]; unsigned char buf[4];
int len = MQTTSNSerialize_unsuback(buf, 4, (unsigned short) msgId); int buflen = sizeof(buf);
int len = MQTTSNSerialize_unsuback(buf, buflen, (unsigned short) msgId);
return desirialize(buf, len); return desirialize(buf, len);
} }
int MQTTSNPacket::setPINGRESP(void) int MQTTSNPacket::setPINGRESP(void)
{ {
unsigned char buf[32]; unsigned char buf[32];
int len = MQTTSNSerialize_pingresp(buf, 32); int buflen = sizeof(buf);
int len = MQTTSNSerialize_pingresp(buf, buflen);
return desirialize(buf, len); return desirialize(buf, len);
} }
int MQTTSNPacket::setDISCONNECT(uint16_t duration) int MQTTSNPacket::setDISCONNECT(uint16_t duration)
{ {
unsigned char buf[4]; unsigned char buf[4];
int len = MQTTSNSerialize_disconnect(buf, 4, (int) duration); int buflen = sizeof(buf);
int len = MQTTSNSerialize_disconnect(buf, buflen, (int) duration);
return desirialize(buf, len); return desirialize(buf, len);
} }
int MQTTSNPacket::setWILLTOPICRESP(uint8_t returnCode) int MQTTSNPacket::setWILLTOPICRESP(uint8_t returnCode)
{ {
unsigned char buf[MQTTSNGW_MAX_PACKET_SIZE]; unsigned char buf[MQTTSNGW_MAX_PACKET_SIZE];
int len = MQTTSNSerialize_willtopicresp(buf, MQTTSNGW_MAX_PACKET_SIZE, (int) returnCode); int buflen = sizeof(buf);
int len = MQTTSNSerialize_willtopicresp(buf, buflen, (int) returnCode);
return desirialize(buf, len); return desirialize(buf, len);
} }
int MQTTSNPacket::setWILLMSGRESP(uint8_t returnCode) int MQTTSNPacket::setWILLMSGRESP(uint8_t returnCode)
{ {
unsigned char buf[MQTTSNGW_MAX_PACKET_SIZE]; unsigned char buf[MQTTSNGW_MAX_PACKET_SIZE];
int len = MQTTSNSerialize_willmsgresp(buf, MQTTSNGW_MAX_PACKET_SIZE, (int) returnCode); int buflen = sizeof(buf);
int len = MQTTSNSerialize_willmsgresp(buf, buflen, (int) returnCode);
return desirialize(buf, len);
}
int MQTTSNPacket::setCONNECT(MQTTSNPacket_connectData* options)
{
unsigned char buf[MQTTSNGW_MAX_PACKET_SIZE];
int buflen = sizeof(buf);
int len = MQTTSNSerialize_connect(buf, buflen, options);
return desirialize(buf, len);
}
int MQTTSNPacket::setPINGREQ(MQTTSNString* clientId)
{
unsigned char buf[MQTTSNGW_MAX_PACKET_SIZE];
int buflen = sizeof(buf);
int len = MQTTSNSerialize_pingreq( buf, buflen, *clientId);
return desirialize(buf, len); return desirialize(buf, len);
} }

View File

@@ -59,6 +59,9 @@ public:
int setWILLTOPICRESP(uint8_t returnCode); int setWILLTOPICRESP(uint8_t returnCode);
int setWILLMSGRESP(uint8_t returnCode); int setWILLMSGRESP(uint8_t returnCode);
int setCONNECT(MQTTSNPacket_connectData* options);
int setPINGREQ(MQTTSNString* clientId);
int getSERCHGW(uint8_t* radius); int getSERCHGW(uint8_t* radius);
int getCONNECT(MQTTSNPacket_connectData* option); int getCONNECT(MQTTSNPacket_connectData* option);
int getCONNACK(uint8_t* returnCode); int getCONNACK(uint8_t* returnCode);
@@ -76,6 +79,8 @@ public:
int getDISCONNECT(uint16_t* duration); int getDISCONNECT(uint16_t* duration);
int getWILLTOPICUPD(uint8_t* willQoS, uint8_t* willRetain, MQTTSNString* willTopic); int getWILLTOPICUPD(uint8_t* willQoS, uint8_t* willRetain, MQTTSNString* willTopic);
int getWILLMSGUPD(MQTTSNString* willMsg); int getWILLMSGUPD(MQTTSNString* willMsg);
bool isQoSMinusPUBLISH(void);
char* getMsgId(char* buf); char* getMsgId(char* buf);
char* print(char* buf); char* print(char* buf);

View File

@@ -113,6 +113,9 @@ void PacketHandleTask::run()
_mqttsnConnection->sendADVERTISE(); _mqttsnConnection->sendADVERTISE();
_advertiseTimer.start(_gateway->getGWParams()->keepAlive * 1000UL); _advertiseTimer.start(_gateway->getGWParams()->keepAlive * 1000UL);
} }
/*------ Check ClientProxy to Connect or send PINGREQ ------*/
_gateway->getClientProxy()->checkConnection();
} }
/*------ Handle SEARCHGW Message ---------*/ /*------ Handle SEARCHGW Message ---------*/

View File

@@ -43,10 +43,17 @@ void MQTTSNPublishHandler::handlePublish(Client* client, MQTTSNPacket* packet)
uint8_t* payload; uint8_t* payload;
MQTTSN_topicid topicid; MQTTSN_topicid topicid;
int payloadlen; int payloadlen;
Publish pub; Publish pub = {0, 0, 0, 0, 0, 0};
char shortTopic[2]; char shortTopic[2];
if ( !client->isActive() ) if ( !client->isActive() )
{
if ( client->isProxy() )
{
client->setProxyPacket(packet);
}
else
{ {
/* Reply DISCONNECT to the client */ /* Reply DISCONNECT to the client */
Event* ev = new Event(); Event* ev = new Event();
@@ -54,6 +61,7 @@ void MQTTSNPublishHandler::handlePublish(Client* client, MQTTSNPacket* packet)
disconnect->setDISCONNECT(0); disconnect->setDISCONNECT(0);
ev->setClientSendEvent(client, disconnect); ev->setClientSendEvent(client, disconnect);
_gateway->getClientSendQue()->post(ev); _gateway->getClientSendQue()->post(ev);
}
return; return;
} }
@@ -63,7 +71,7 @@ void MQTTSNPublishHandler::handlePublish(Client* client, MQTTSNPacket* packet)
} }
pub.msgId = msgId; pub.msgId = msgId;
pub.header.bits.dup = dup; pub.header.bits.dup = dup;
pub.header.bits.qos = qos; pub.header.bits.qos = ( qos == 3 ? 0 : qos );
pub.header.bits.retain = retained; pub.header.bits.retain = retained;
Topic* topic = 0; Topic* topic = 0;
@@ -79,7 +87,13 @@ void MQTTSNPublishHandler::handlePublish(Client* client, MQTTSNPacket* packet)
{ {
topic = client->getTopics()->getTopicById(&topicid); topic = client->getTopics()->getTopicById(&topicid);
if( !topic && msgId && qos > 0 ) if( !topic && qos == 3 )
{
WRITELOG("%s Invali TopicId.%s\n", ERRMSG_HEADER, client->getClientId(), ERRMSG_FOOTER);
return;
}
if( !topic && msgId && qos > 0 && qos < 3 )
{ {
/* Reply PubAck with INVALID_TOPIC_ID to the client */ /* Reply PubAck with INVALID_TOPIC_ID to the client */
MQTTSNPacket* pubAck = new MQTTSNPacket(); MQTTSNPacket* pubAck = new MQTTSNPacket();
@@ -96,7 +110,7 @@ void MQTTSNPublishHandler::handlePublish(Client* client, MQTTSNPacket* packet)
} }
} }
/* Save a msgId & a TopicId pare for PUBACK */ /* Save a msgId & a TopicId pare for PUBACK */
if( msgId && qos > 0 ) if( msgId && qos > 0 && qos < 3)
{ {
client->setWaitedPubTopicId(msgId, topicid.data.id, topicid.type); client->setWaitedPubTopicId(msgId, topicid.data.id, topicid.type);
} }
@@ -161,7 +175,7 @@ void MQTTSNPublishHandler::handleRegister(Client* client, MQTTSNPacket* packet)
{ {
uint16_t id; uint16_t id;
uint16_t msgId; uint16_t msgId;
MQTTSNString topicName; MQTTSNString topicName = MQTTSNString_initializer;;
MQTTSN_topicid topicid; MQTTSN_topicid topicid;
if ( client->isActive() || client->isAwake()) if ( client->isActive() || client->isAwake())

View File

@@ -30,6 +30,7 @@ Gateway::Gateway()
{ {
theMultiTaskProcess = this; theMultiTaskProcess = this;
theProcess = this; theProcess = this;
_clientProxy = new ClientProxy(this);
_params.loginId = 0; _params.loginId = 0;
_params.password = 0; _params.password = 0;
_params.keepAlive = 0; _params.keepAlive = 0;
@@ -48,6 +49,7 @@ Gateway::Gateway()
_params.configName = 0; _params.configName = 0;
_params.predefinedTopicFileName = 0; _params.predefinedTopicFileName = 0;
_params.forwarderListName = 0; _params.forwarderListName = 0;
_params.qosMinusClientListName = 0;
_packetEventQue.setMaxSize(MAX_INFLIGHTMESSAGES * MAX_CLIENTS); _packetEventQue.setMaxSize(MAX_INFLIGHTMESSAGES * MAX_CLIENTS);
} }
@@ -109,12 +111,21 @@ Gateway::~Gateway()
{ {
free(_params.forwarderListName); free(_params.forwarderListName);
} }
if ( _params.qosMinusClientListName )
{
free(_params.qosMinusClientListName);
}
if ( _clientProxy )
{
delete _clientProxy;
}
} }
void Gateway::initialize(int argc, char** argv) void Gateway::initialize(int argc, char** argv)
{ {
char param[MQTTSNGW_PARAM_MAX]; char param[MQTTSNGW_PARAM_MAX];
string fileName; string fileName;
bool secure = false;
MultiTaskProcess::initialize(argc, argv); MultiTaskProcess::initialize(argc, argv);
resetRingBuffer(); resetRingBuffer();
@@ -201,6 +212,7 @@ void Gateway::initialize(int argc, char** argv)
{ {
if (!strcasecmp(param, "YES")) if (!strcasecmp(param, "YES"))
{ {
secure = true;
if (getParam("ClientsList", param) == 0) if (getParam("ClientsList", param) == 0)
{ {
fileName = string(param); fileName = string(param);
@@ -218,7 +230,33 @@ void Gateway::initialize(int argc, char** argv)
} }
} }
/* Set ClientProxy's Client */
MQTTSNString id = MQTTSNString_initializer;
id.cstring = const_cast<char*>(CLIENTPROXY);
Client* client = _clientList.createClient(0, &id, true, secure);
_clientProxy->setClient(client);
client->setPorxy(true);
_clientProxy->setGateway(this);
if (getParam("QoS-1", param) == 0 )
{
if (!strcasecmp(param, "YES") )
{
if (getParam("QoS-1ClientsList", param) == 0)
{
fileName = string(param);
}
else
{
fileName = *getConfigDirName() + string(QOS_1CLIENT_LIST);
}
if ( !_clientProxy->setClientProxy(fileName.c_str()) )
{
throw Exception("Gateway::initialize: No QoS-1ClientsList file defined by the configuration..");
}
_params.qosMinusClientListName = strdup(fileName.c_str());
}
}
if (getParam("PredefinedTopic", param) == 0 ) if (getParam("PredefinedTopic", param) == 0 )
{ {
@@ -238,10 +276,6 @@ void Gateway::initialize(int argc, char** argv)
} }
_params.predefinedTopicFileName = strdup(fileName.c_str()); _params.predefinedTopicFileName = strdup(fileName.c_str());
} }
else
{
_params.predefinedTopicFileName = 0;
}
} }
if (getParam("Forwarder", param) == 0 ) if (getParam("Forwarder", param) == 0 )
@@ -262,14 +296,11 @@ void Gateway::initialize(int argc, char** argv)
} }
_params.forwarderListName = strdup(fileName.c_str()); _params.forwarderListName = strdup(fileName.c_str());
} }
else
{
_params.forwarderListName = 0;
}
} }
fileName = *getConfigDirName() + *getConfigFileName(); fileName = *getConfigDirName() + *getConfigFileName();
_params.configName = strdup(fileName.c_str()); _params.configName = strdup(fileName.c_str());
} }
void Gateway::run(void) void Gateway::run(void)
@@ -295,6 +326,10 @@ void Gateway::run(void)
if ( _params.forwarderListName ) if ( _params.forwarderListName )
{ {
WRITELOG(" Forwarders: %s\n", _params.forwarderListName); WRITELOG(" Forwarders: %s\n", _params.forwarderListName);
}
if ( _params.qosMinusClientListName )
{
WRITELOG(" QoS-1: %s\n", _params.qosMinusClientListName);
} }
WRITELOG(" SensorN/W: %s\n", _sensorNetwork.getDescription()); WRITELOG(" SensorN/W: %s\n", _sensorNetwork.getDescription());
WRITELOG(" Broker: %s : %s, %s\n", _params.brokerName, _params.port, _params.portSecure); WRITELOG(" Broker: %s : %s, %s\n", _params.brokerName, _params.port, _params.portSecure);
@@ -302,7 +337,7 @@ void Gateway::run(void)
WRITELOG(" RootCApath: %s\n", _params.rootCApath); WRITELOG(" RootCApath: %s\n", _params.rootCApath);
WRITELOG(" RootCAfile: %s\n", _params.rootCAfile); WRITELOG(" RootCAfile: %s\n", _params.rootCAfile);
WRITELOG(" CertKey: %s\n", _params.certKey); WRITELOG(" CertKey: %s\n", _params.certKey);
WRITELOG(" PrivateKey: %s\n", _params.privateKey); WRITELOG(" PrivateKey: %s\n\n\n", _params.privateKey);
MultiTaskProcess::run(); MultiTaskProcess::run();
@@ -364,6 +399,11 @@ GatewayParams* Gateway::getGWParams(void)
return &_params; return &_params;
} }
ClientProxy* Gateway::getClientProxy(void)
{
return _clientProxy;
}
/*===================================== /*=====================================
Class EventQue Class EventQue
=====================================*/ =====================================*/

View File

@@ -17,10 +17,12 @@
#define MQTTSNGATEWAY_H_ #define MQTTSNGATEWAY_H_
#include "MQTTSNGWClient.h" #include "MQTTSNGWClient.h"
#include "MQTTSNGWForwarder.h"
#include "MQTTSNGWProcess.h" #include "MQTTSNGWProcess.h"
#include "MQTTSNPacket.h" #include "MQTTSNPacket.h"
#include "MQTTSNGWForwarder.h"
#include "MQTTSNGWClientProxy.h"
namespace MQTTSNGW namespace MQTTSNGW
{ {
/*================================= /*=================================
@@ -43,6 +45,7 @@ namespace MQTTSNGW
#define CLIENT "Client" #define CLIENT "Client"
#define CLIENTS "Clients" #define CLIENTS "Clients"
#define UNKNOWNCL "Unknown Client !" #define UNKNOWNCL "Unknown Client !"
#define CLIENTPROXY "ClientProxy"
#define LEFTARROW "<---" #define LEFTARROW "<---"
#define RIGHTARROW "--->" #define RIGHTARROW "--->"
@@ -156,11 +159,14 @@ typedef struct
char* privateKey; char* privateKey;
char* predefinedTopicFileName; char* predefinedTopicFileName;
char* forwarderListName; char* forwarderListName;
char* qosMinusClientListName;
}GatewayParams; }GatewayParams;
/*===================================== /*=====================================
Class Gateway Class Gateway
=====================================*/ =====================================*/
class ClientProxy;
class Gateway: public MultiTaskProcess{ class Gateway: public MultiTaskProcess{
public: public:
Gateway(); Gateway();
@@ -176,9 +182,11 @@ public:
SensorNetwork* getSensorNetwork(void); SensorNetwork* getSensorNetwork(void);
LightIndicator* getLightIndicator(void); LightIndicator* getLightIndicator(void);
GatewayParams* getGWParams(void); GatewayParams* getGWParams(void);
ClientProxy* getClientProxy(void);
private: private:
ClientList _clientList; ClientList _clientList;
ClientProxy* _clientProxy;
ForwarderList _forwarderList; ForwarderList _forwarderList;
EventQue _packetEventQue; EventQue _packetEventQue;
EventQue _brokerSendQue; EventQue _brokerSendQue;