Merge pull request #217 from eclipse/develop

Merge latest develop branch to master

.cproject

@@ -53,17 +53,17 @@
                                 								
                                 <option IS_BUILTIN_EMPTY="false" IS_VALUE_EMPTY="false" id="gnu.cpp.compiler.option.include.paths.1361987608" name="Include paths (-I)" superClass="gnu.cpp.compiler.option.include.paths" useByScannerDiscovery="false" valueType="includePath">
                                     									
-                                    <listOptionValue builtIn="false" value="/home/tomoaki/git/paho.mqtt-sn.embedded-c/MQTTSNGateway/src"/>
+                                    <listOptionValue builtIn="false" value="&quot;${workspace_loc:/${ProjName}/MQTTSNGateway/src}&quot;"/>
                                     									
-                                    <listOptionValue builtIn="false" value="/home/tomoaki/git/paho.mqtt-sn.embedded-c/MQTTSNGateway/GatewayTester/src"/>
+                                    <listOptionValue builtIn="false" value="&quot;${workspace_loc:/${ProjName}/MQTTSNGateway/GatewayTester/src}&quot;"/>
                                     									
-                                    <listOptionValue builtIn="false" value="/home/tomoaki/git/paho.mqtt-sn.embedded-c/MQTTSNGateway/src/linux"/>
+                                    <listOptionValue builtIn="false" value="&quot;${workspace_loc:/${ProjName}/MQTTSNGateway/src/linux}&quot;"/>
                                     									
-                                    <listOptionValue builtIn="false" value="/home/tomoaki/git/paho.mqtt-sn.embedded-c/MQTTSNPacket/src"/>
+                                    <listOptionValue builtIn="false" value="&quot;${workspace_loc:/${ProjName}/MQTTSNPacket/src}&quot;"/>
                                     									
-                                    <listOptionValue builtIn="false" value="/home/tomoaki/git/paho.mqtt-sn.embedded-c/MQTTSNClient/src"/>
+                                    <listOptionValue builtIn="false" value="&quot;${workspace_loc:/${ProjName}/MQTTSNClient/src}&quot;"/>
                                     									
-                                    <listOptionValue builtIn="false" value="/home/tomoaki/git/paho.mqtt-sn.embedded-c/MQTTSNGateway/src/linux/udp"/>
+                                    <listOptionValue builtIn="false" value="&quot;${workspace_loc:/${ProjName}/MQTTSNGateway/src/linux/udp}&quot;"/>
                                     								
                                 </option>
                                 								
@@ -133,7 +133,7 @@
                     					
                     <sourceEntries>
                         						
-                        <entry excluding="MQTTSNGateway/src/linux/udp6|MQTTSNGateway/GatewayTester|MQTTSNClient|MQTTSNGateway/src/MQTTSNGWProxy.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|ClientPubQoS-1|MQTTSNGateway/GatewayTester/samples/mainOTA.cpp" flags="VALUE_WORKSPACE_PATH|RESOLVED" kind="sourcePath" name=""/>
+                        <entry excluding="MQTTSNGateway/src/linux/loralink|MQTTSNGateway/GatewayTester|MQTTSNGateway/GatewayTester/samples|MQTTSNGateway/src/linux/udp6|MQTTSNClient|MQTTSNGateway/src/MQTTSNGWProxy.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|ClientPubQoS-1|MQTTSNGateway/GatewayTester/samples/mainOTA.cpp" flags="VALUE_WORKSPACE_PATH" kind="sourcePath" name=""/>
                         					
                     </sourceEntries>
                     				
@@ -193,15 +193,15 @@
                                 								
                                 <option IS_BUILTIN_EMPTY="false" IS_VALUE_EMPTY="false" id="gnu.cpp.compiler.option.include.paths.2114194326" name="Include paths (-I)" superClass="gnu.cpp.compiler.option.include.paths" useByScannerDiscovery="false" valueType="includePath">
                                     									
-                                    <listOptionValue builtIn="false" value="/home/tomoaki/git/paho.mqtt-sn.embedded-c/MQTTSNGateway/src/linux/udp"/>
+                                    <listOptionValue builtIn="false" value="&quot;${workspace_loc:/${ProjName}/MQTTSNGateway/src/linux/udp}&quot;"/>
                                     									
-                                    <listOptionValue builtIn="false" value="/home/tomoaki/git/paho.mqtt-sn.embedded-c/MQTTSNGateway/GatewayTester/src"/>
+                                    <listOptionValue builtIn="false" value="&quot;${workspace_loc:/${ProjName}/MQTTSNPacket/src}&quot;"/>
                                     									
-                                    <listOptionValue builtIn="false" value="/home/tomoaki/git/paho.mqtt-sn.embedded-c/MQTTSNGateway/src"/>
+                                    <listOptionValue builtIn="false" value="&quot;${workspace_loc:/${ProjName}/MQTTSNGateway/src/linux}&quot;"/>
                                     									
-                                    <listOptionValue builtIn="false" value="/home/tomoaki/git/paho.mqtt-sn.embedded-c/MQTTSNGateway/src/linux"/>
+                                    <listOptionValue builtIn="false" value="&quot;${workspace_loc:/${ProjName}/MQTTSNGateway/src}&quot;"/>
                                     									
-                                    <listOptionValue builtIn="false" value="/home/tomoaki/git/paho.mqtt-sn.embedded-c/MQTTSNPacket/src"/>
+                                    <listOptionValue builtIn="false" value="&quot;${workspace_loc:/${ProjName}/MQTTSNGateway/GatewayTester/src}&quot;"/>
                                     								
                                 </option>
                                 								
@@ -273,11 +273,9 @@
                     					
                     <sourceEntries>
                         						
-                        <entry excluding="MQTTSNGateway/GatewayTester|MQTTSNClient|MQTTSNGateway/src/linux/udp|MQTTSNPacket/test|MQTTSNPacket/samples|MQTTSNGateway/src/linux|MQTTSNGateway/GatewayTester/samples/mainOTA.cpp|MQTTSNGateway/src/mainLogmonitor.cpp|MQTTSNGateway/GatewayTester/samples/mainTemplate.cpp|MQTTSNGateway/src/linux/xbee|MQTTSNGateway/src|MQTTSNPacket/src|ClientPubQoS-1" flags="VALUE_WORKSPACE_PATH|RESOLVED" kind="sourcePath" name=""/>
+                        <entry excluding="MQTTSNGateway/src/tests/mainTestProcess.cpp|MQTTSNGateway/src/linux|MQTTSNPacket/src|MQTTSNGateway/GatewayTester|MQTTSNGateway/GatewayTester/samples|MQTTSNClient|MQTTSNPacket/test|MQTTSNPacket/samples|MQTTSNGateway/GatewayTester/samples/mainOTA.cpp|MQTTSNGateway/src/mainLogmonitor.cpp|MQTTSNGateway/GatewayTester/samples/mainTemplate.cpp|ClientPubQoS-1" flags="VALUE_WORKSPACE_PATH" kind="sourcePath" name=""/>
                         						
-                        <entry excluding="MQTTSNGWProxy.cpp|mainLogmonitor.cpp|tests|tests/mainTestProcessFramework.cpp|linux|linux/udp" flags="VALUE_WORKSPACE_PATH" kind="sourcePath" name="MQTTSNGateway/src"/>
-                        						
-                        <entry excluding="udp6|xbee" flags="VALUE_WORKSPACE_PATH" kind="sourcePath" name="MQTTSNGateway/src/linux"/>
+                        <entry excluding="udp6|xbee|loralink" flags="VALUE_WORKSPACE_PATH|RESOLVED" kind="sourcePath" name="MQTTSNGateway/src/linux"/>
                         						
                         <entry flags="VALUE_WORKSPACE_PATH" kind="sourcePath" name="MQTTSNPacket/src"/>
                         					
@@ -319,17 +317,19 @@
     	
     <storageModule moduleId="org.eclipse.cdt.internal.ui.text.commentOwnerProjectMappings"/>
     	
+    <storageModule moduleId="org.eclipse.cdt.make.core.buildtargets"/>
+    	
     <storageModule moduleId="scannerConfiguration">
         		
         <autodiscovery enabled="true" problemReportingEnabled="true" selectedProfileId=""/>
         		
-        <scannerConfigBuildInfo instanceId="cdt.managedbuild.config.gnu.exe.release.561557339;cdt.managedbuild.config.gnu.exe.release.561557339.;cdt.managedbuild.tool.gnu.cpp.compiler.exe.release.1903732701;cdt.managedbuild.tool.gnu.cpp.compiler.input.1606625536">
+        <scannerConfigBuildInfo instanceId="cdt.managedbuild.config.gnu.exe.debug.1685199227;cdt.managedbuild.config.gnu.exe.debug.1685199227.;cdt.managedbuild.tool.gnu.c.compiler.exe.debug.2017162618;cdt.managedbuild.tool.gnu.c.compiler.input.814497727">
             			
             <autodiscovery enabled="true" problemReportingEnabled="true" selectedProfileId=""/>
             		
         </scannerConfigBuildInfo>
         		
-        <scannerConfigBuildInfo instanceId="cdt.managedbuild.config.gnu.exe.debug.1685199227;cdt.managedbuild.config.gnu.exe.debug.1685199227.;cdt.managedbuild.tool.gnu.c.compiler.exe.debug.2017162618;cdt.managedbuild.tool.gnu.c.compiler.input.814497727">
+        <scannerConfigBuildInfo instanceId="cdt.managedbuild.config.gnu.exe.release.561557339;cdt.managedbuild.config.gnu.exe.release.561557339.;cdt.managedbuild.tool.gnu.cpp.compiler.exe.release.1903732701;cdt.managedbuild.tool.gnu.cpp.compiler.input.1606625536">
             			
             <autodiscovery enabled="true" problemReportingEnabled="true" selectedProfileId=""/>
             		
@@ -348,7 +348,5 @@
         </scannerConfigBuildInfo>
         	
     </storageModule>
-    	
-    <storageModule moduleId="org.eclipse.cdt.make.core.buildtargets"/>
     
 </cproject>

.settings/language.settings.xml

@@ -11,7 +11,7 @@
             			
             <provider-reference id="org.eclipse.cdt.managedbuilder.core.MBSLanguageSettingsProvider" ref="shared-provider"/>
             			
-            <provider class="org.eclipse.cdt.managedbuilder.language.settings.providers.GCCBuiltinSpecsDetector" console="false" env-hash="-181038896859540458" id="org.eclipse.cdt.managedbuilder.core.GCCBuiltinSpecsDetector" keep-relative-paths="false" name="CDT GCC Built-in Compiler Settings" parameter="${COMMAND} ${FLAGS} -E -P -v -dD &quot;${INPUTS}&quot;" prefer-non-shared="true">
+            <provider class="org.eclipse.cdt.managedbuilder.language.settings.providers.GCCBuiltinSpecsDetector" console="false" env-hash="-179845344490928458" id="org.eclipse.cdt.managedbuilder.core.GCCBuiltinSpecsDetector" keep-relative-paths="false" name="CDT GCC Built-in Compiler Settings" parameter="${COMMAND} ${FLAGS} -E -P -v -dD &quot;${INPUTS}&quot;" prefer-non-shared="true">
                 				
                 <language-scope id="org.eclipse.cdt.core.gcc"/>
                 				
@@ -33,7 +33,7 @@
             			
             <provider-reference id="org.eclipse.cdt.managedbuilder.core.MBSLanguageSettingsProvider" ref="shared-provider"/>
             			
-            <provider class="org.eclipse.cdt.managedbuilder.language.settings.providers.GCCBuiltinSpecsDetector" console="false" env-hash="-181038896859540458" id="org.eclipse.cdt.managedbuilder.core.GCCBuiltinSpecsDetector" keep-relative-paths="false" name="CDT GCC Built-in Compiler Settings" parameter="${COMMAND} ${FLAGS} -E -P -v -dD &quot;${INPUTS}&quot;" prefer-non-shared="true">
+            <provider class="org.eclipse.cdt.managedbuilder.language.settings.providers.GCCBuiltinSpecsDetector" console="false" env-hash="-179845344490928458" id="org.eclipse.cdt.managedbuilder.core.GCCBuiltinSpecsDetector" keep-relative-paths="false" name="CDT GCC Built-in Compiler Settings" parameter="${COMMAND} ${FLAGS} -E -P -v -dD &quot;${INPUTS}&quot;" prefer-non-shared="true">
                 				
                 <language-scope id="org.eclipse.cdt.core.gcc"/>
                 				

MQTTSNGateway/GatewayTester/src/LPublishManager.cpp

@@ -119,7 +119,7 @@ void LPublishManager::sendPublish(PubElement* elm)
 
     uint8_t msg[MQTTSN_MAX_MSG_LENGTH + 1];
     uint8_t org = 0;
-    if (elm->payloadlen > 128)
+    if (elm->payloadlen > 248)
     {
         msg[0] = 0x01;
         setUint16(msg + 1, elm->payloadlen + 9);

MQTTSNGateway/GatewayTester/src/LTaskManager.cpp

@@ -56,12 +56,14 @@ void LTaskManager::run(void){
 	int i = 0;
 	char c = 0;
 	bool cancelFlg = false;
+	TestList test = {0};
+	TaskList task = {0};
 
 	if ( !theClientMode )
 	{
 			theClient->getGwProxy()->getMessage();
 
-			for (i = 0; _tests[i].testTask > 0; i++)
+			for (i = 0; _tests[i].testTask > test.testTask; i++)
 			{
 				PROMPT("Execute \"%s\" ? ( y/n ) : ", _tests[i].testLabel);
 				while (true)
@@ -118,7 +120,7 @@ void LTaskManager::run(void){
 		while (true)
 		{
 			theClient->getGwProxy()->getMessage();
-			for (_index = 0; _tasks[_index].callback > 0; _index++)
+			for (_index = 0; _tasks[_index].callback > task.callback; _index++)
 			{
 				if ((_tasks[_index].prevTime + _tasks[_index].interval <= time(NULL)) &&
 					 _tasks[_index].count == 0)

MQTTSNGateway/GatewayTester/src/LTimer.h

@@ -17,7 +17,7 @@
 #ifndef TIMER_H_
 #define TIMER_H_
 
-#include <time.h>
+#include <sys/time.h>
 
 #include "LMqttsnClientApp.h"
 

MQTTSNGateway/Makefile

@@ -74,7 +74,6 @@ $(SUBDIR)/MQTTSNSubscribeServer.c \
 $(SUBDIR)/MQTTSNUnsubscribeClient.c \
 $(SUBDIR)/MQTTSNUnsubscribeServer.c 
 
-CXX := g++
 CPPFLAGS += 
 
 INCLUDE :=
@@ -82,14 +81,20 @@ INCLUDES += $(INCLUDE) -I$(SRCDIR) \
 -I$(SRCDIR)/$(OS) \
 -I$(SRCDIR)/$(OS)/$(SENSORNET) \
 -I$(SUBDIR) \
--I$(SRCDIR)/$(TEST)
+-I$(SRCDIR)/$(TEST) \
+-I/usr/local/opt/openssl/include/
 
+# preprocessor defines
 DEFS :=
+
+CXX := g++
+
 LIB :=
-LIBS += $(LIB) -L/usr/local/lib
-LDFLAGS := 
+LIBS += $(LIB) -L/usr/local/lib -L/usr/local/opt/openssl/lib/
+
+LDFLAGS :=
 CXXFLAGS := -Wall -O3 -std=c++11
-LDADD := -lpthread -lssl -lcrypto -lrt
+LDADD := -lpthread -lssl -lcrypto
 OUTDIR := Build
 
 PROG := $(OUTDIR)/$(PROGNAME)

MQTTSNGateway/README.md

@@ -1,18 +1,18 @@
-# MQTT-SN Transparent / Aggrigating Gateway
+# MQTT-SN Transparent / Aggregating Gateway
 
 **MQTT-SN** requires a MQTT-SN Gateway which acts as a protocol converter to convert **MQTT-SN messages to MQTT messages**. MQTT-SN client over SensorNetwork can not communicate directly with MQTT broker(TCP/IP).   
-This Gateway can run as a transparent or aggrigating Gateway by specifying the gateway.conf.
+This Gateway can run as a transparent or aggregating Gateway by specifying the gateway.conf.    
 
 ### **step1. Build the gateway**   
 ````
-$ git clone -b experiment https://github.com/eclipse/paho.mqtt-sn.embedded-c   
+$ git clone -b develop https://github.com/eclipse/paho.mqtt-sn.embedded-c   
 $ cd paho.mqtt-sn.embedded-c/MQTTSNGateway       
-$ make [SENSORNET={udp6|xbee}] 
+$ make [SENSORNET={udp6|xbee|loralink}] 
 $ make install   
 $ make clean    
 ````      
 By default, a gateway for UDP is built.    
-In order to create a gateway for UDP6 or XBee, SENSORNET argument is required.  
+In order to create a gateway for UDP6, XBee or LoRaLink, SENSORNET argument is required.  
  
 MQTT-SNGateway, MQTT-SNLogmonitor and *.conf files are copied into ../ directory.    
 If you want to install the gateway into specific directories, enter a command line as follows:
@@ -24,10 +24,18 @@ $ make install INSTALL_DIR=/path/to/your_directory CONFIG_DIR=/path/to/your_dire
 ### **step2. Execute the Gateway.**     
 
 ````    
-$ cd ../   
+$ cd ../../   
 $ ./MQTT-SNGateway [-f Config file name]
 ````   
-
+If you get the error message as follows:
+````    
+what(): RingBuffer can't create a shared memory.
+Aborted (core dumped)
+````
+You have to start using sudo command only once for the first time.    
+````
+$ sudo ./MQTT-SNGateway [-f Config file name]
+````
 
 ### **How to Change the configuration of the gateway**    
 **../gateway.conf**   Contents are follows: 
@@ -37,7 +45,7 @@ $ ./MQTT-SNGateway [-f Config file name]
 # config file of MQTT-SN Gateway
 #
 
-BrokerName=iot.eclipse.org
+BrokerName=mqtt.eclipse.org
 BrokerPortNo=1883
 BrokerSecurePortNo=8883
 
@@ -72,12 +80,25 @@ KeepAlive=900
 GatewayPortNo=10000
 MulticastIP=225.1.1.1
 MulticastPortNo=1883
+MulticastTTL=1  
+
+# UDP6
+GatewayUDP6Bind=FFFF:FFFE::1 
+GatewayUDP6Port=10000
+GatewayUDP6Broadcast=FF02::1
+GatewayUDP6If=wpan0
+GatewayUDP6Hops=1
 
 # XBee
 Baudrate=38400
 SerialDevice=/dev/ttyUSB0
 ApiMode=2
 
+#LoRaLink
+BaudrateLoRaLink=115200
+DeviceRxLoRaLink=/dev/ttyLoRaLinkRx
+DeviceTxLoRaLink=/dev/ttyLoRaLinkTx
+
 # LOG
 ShearedMemory=NO;
 

MQTTSNGateway/gateway.conf

@@ -49,12 +49,14 @@ KeepAlive=900
 GatewayPortNo=10000
 MulticastIP=225.1.1.1
 MulticastPortNo=1883
+MulticastTTL=1
 
 # UDP6
 GatewayUDP6Bind=FFFF:FFFE::1 
 GatewayUDP6Port=10000
 GatewayUDP6Broadcast=FF02::1
 GatewayUDP6If=wpan0
+GatewayUDP6Hops=1
 
 # XBee
 Baudrate=38400

MQTTSNGateway/src/MQTTGWPacket.h

@@ -116,7 +116,7 @@ typedef struct
 	unsigned char version;	/**< MQTT version number */
 } Connect;
 
-#define MQTTPacket_Connect_Initializer {{0}, 0, nullptr, nullptr, nullptr, nullptr, 0, 0}
+#define MQTTPacket_Connect_Initializer {{0}, {0}, nullptr, nullptr, nullptr, nullptr, 0, 0}
 #define MQTTPacket_willOptions_initializer { {'M', 'Q', 'T', 'W'}, 0, {NULL, {0, NULL}}, {NULL, {0, NULL}}, 0, 0 }
 #define MQTTPacket_connectData_initializer { {'M', 'Q', 'T', 'C'}, 0, 4, {NULL, {0, NULL}}, 60, 1, 0, \
         MQTTPacket_willOptions_initializer, {NULL, {0, NULL}}, {NULL, {0, NULL}} }

MQTTSNGateway/src/MQTTGWPublishHandler.cpp

@@ -272,52 +272,31 @@ void MQTTGWPublishHandler::handleAggregatePublish(Client* client, MQTTGWPacket*
 	Publish pub;
 	packet->getPUBLISH(&pub);
 
-	WRITELOG(FORMAT_Y_G_G, currentDateTime(), packet->getName(),
-	RIGHTARROW, client->getClientId(), "is sleeping. a message was saved.");
 
-	if (pub.header.bits.qos == 1)
-	{
-		replyACK(client, &pub, PUBACK);
-	}
-	else if ( pub.header.bits.qos == 2)
-	{
-		replyACK(client, &pub, PUBREC);
-	}
-
-
-
-	string* topicName = new string(pub.topic, pub.topiclen);
+	string* topicName = new string(pub.topic, pub.topiclen);   // topic deletes topicName when the topic is deleted
 	Topic topic = Topic(topicName, MQTTSN_TOPIC_TYPE_NORMAL);
-	AggregateTopicElement* list = _gateway->getAdapterManager()->createClientList(&topic);
-	if ( list != nullptr )
+
+	// ToDo: need to refactor
+	ClientTopicElement* elm = _gateway->getAdapterManager()->getAggregater()->getClientElement(&topic);
+
+	while ( elm != nullptr )
 	{
-		ClientTopicElement* p = list->getFirstElement();
+		Client* devClient = elm->getClient();
+		MQTTGWPacket* msg = new MQTTGWPacket();
+		*msg = *packet;
 
-		while ( p )
+		if ( msg->getType() == 0 )
 		{
-			Client* devClient = p->getClient();
-			if ( devClient != nullptr )
-			{
-				MQTTGWPacket* msg = new MQTTGWPacket();
-				*msg = *packet;
-				if ( msg->getType() == 0 )
-				{
-					WRITELOG("%s MQTTGWPublishHandler::handleAggregatePublish can't allocate memories for Packet.%s\n", ERRMSG_HEADER,ERRMSG_FOOTER);
-					delete msg;
-					break;
-				}
-				Event* ev = new Event();
-				ev->setBrokerRecvEvent(devClient, msg);
-				_gateway->getPacketEventQue()->post(ev);
-			}
-			else
-			{
-				break;
-			}
-
-			p = list->getNextElement(p);
+			WRITELOG("%s MQTTGWPublishHandler::handleAggregatePublish can't allocate memories for Packet.%s\n", ERRMSG_HEADER,ERRMSG_FOOTER);
+			delete msg;
+			break;
 		}
-		delete list;
+
+		Event* ev = new Event();
+		ev->setBrokerRecvEvent(devClient, msg);
+		_gateway->getPacketEventQue()->post(ev);
+
+		elm = elm->getNextClientElement();
 	}
 }
 

MQTTSNGateway/src/MQTTSNAggregateConnectionHandler.cpp

@@ -81,7 +81,15 @@ void MQTTSNAggregateConnectionHandler::handleConnect(Client* client, MQTTSNPacke
 		/* renew the TopicList */
 		if (topics)
 		{
-			_gateway->getAdapterManager()->removeAggregateTopicList(topics, client);
+			Topic* tp = topics->getFirstTopic();
+			while( tp != nullptr )
+			{
+				if ( tp->getType() == MQTTSN_TOPIC_TYPE_NORMAL )
+				{
+					_gateway->getAdapterManager()->getAggregater()->removeAggregateTopic(tp, client);
+				}
+				tp = topics->getNextTopic(tp);
+			}
 			topics->eraseNormal();
 		}
 		client->setSessionStatus(true);
@@ -170,19 +178,17 @@ void MQTTSNAggregateConnectionHandler::handlePingreq(Client* client, MQTTSNPacke
 	    sendStoredPublish(client);
 		client->holdPingRequest();
 	}
-	else
-	{
-        /* create and send PINGRESP to the PacketHandler */
-	    client->resetPingRequest();
 
-        MQTTGWPacket* pingresp = new MQTTGWPacket();
+	/* create and send PINGRESP to the PacketHandler */
+	client->resetPingRequest();
 
-        pingresp->setHeader(PINGRESP);
+	MQTTGWPacket* pingresp = new MQTTGWPacket();
 
-        Event* evt = new Event();
-        evt->setBrokerRecvEvent(client, pingresp);
-        _gateway->getPacketEventQue()->post(evt);
-	}
+	pingresp->setHeader(PINGRESP);
+
+	Event* evt = new Event();
+	evt->setBrokerRecvEvent(client, pingresp);
+	_gateway->getPacketEventQue()->post(evt);
 }
 
 void MQTTSNAggregateConnectionHandler::sendStoredPublish(Client* client)
@@ -191,7 +197,6 @@ void MQTTSNAggregateConnectionHandler::sendStoredPublish(Client* client)
 
     while  ( ( msg = client->getClientSleepPacket() ) != nullptr )
     {
-        // ToDo:  This version can't re-send PUBLISH when PUBACK is not returned.
         client->deleteFirstClientSleepPacket();  // pop the que to delete element.
 
         Event* ev = new Event();

MQTTSNGateway/src/MQTTSNAggregateConnectionHandler.h

@@ -39,7 +39,6 @@ public:
 private:
 	void sendStoredPublish(Client* client);
 
-	char _pbuf[MQTTSNGW_MAX_PACKET_SIZE * 3];
 	Gateway* _gateway;
 };
 

MQTTSNGateway/src/MQTTSNGWAdapter.cpp

@@ -126,6 +126,11 @@ bool Adapter::isSecure(SensorNetAddress* addr)
 	}
 }
 
+bool Adapter::isSecure(void)
+{
+	return _isSecure;
+}
+
 void Adapter::setClient(Client* client, bool secure)
 {
     if ( secure )
@@ -212,7 +217,7 @@ Client* Adapter::getAdapterClient(Client* client)
 {
 	if ( client->isSecureNetwork() )
 	{
-		return _client;
+		return _clientSecure;
 	}
 	else
 	{
@@ -241,10 +246,10 @@ void Proxy::checkConnection(Client* client)
     if ( client->isDisconnect()  || ( client->isConnecting() && _responseTimer.isTimeup()) )
     {
         client->connectSended();
-        _responseTimer.start(QOSM1_PROXY_RESPONSE_DURATION * 1000UL);
+        _responseTimer.start(PROXY_RESPONSE_DURATION * 1000UL);
         MQTTSNPacket_connectData options = MQTTSNPacket_connectData_initializer;
         options.clientID.cstring = client->getClientId();
-        options.duration = QOSM1_PROXY_KEEPALIVE_DURATION;
+        options.duration = PROXY_KEEPALIVE_DURATION;
 
         MQTTSNPacket* packet = new MQTTSNPacket();
         packet->setCONNECT(&options);
@@ -260,10 +265,10 @@ void Proxy::checkConnection(Client* client)
             Event* ev = new Event();
             ev->setClientRecvEvent(client, packet);
             _gateway->getPacketEventQue()->post(ev);
-            _responseTimer.start(QOSM1_PROXY_RESPONSE_DURATION * 1000UL);
+            _responseTimer.start(PROXY_RESPONSE_DURATION * 1000UL);
             _isWaitingResp = true;
 
-            if ( ++_retryCnt > QOSM1_PROXY_MAX_RETRY_CNT )
+            if ( ++_retryCnt > PROXY_MAX_RETRY_CNT )
             {
                 client->disconnected();
             }
@@ -274,7 +279,7 @@ void Proxy::checkConnection(Client* client)
 
 void Proxy::resetPingTimer(void)
 {
-    _keepAliveTimer.start(QOSM1_PROXY_KEEPALIVE_DURATION * 1000UL);
+    _keepAliveTimer.start(PROXY_KEEPALIVE_DURATION * 1000UL);
 }
 
 void Proxy::recv(MQTTSNPacket* packet, Client* client)

MQTTSNGateway/src/MQTTSNGWAdapter.h

@@ -56,6 +56,7 @@ public:
     void send(MQTTSNPacket* packet, Client* client);
     bool isActive(void);
     bool isSecure(SensorNetAddress* addr);
+    bool isSecure(void);
     void savePacket(Client* client, MQTTSNPacket* packet);
 
 private:

MQTTSNGateway/src/MQTTSNGWAdapterManager.cpp

@@ -40,13 +40,23 @@ AdapterManager::AdapterManager(Gateway* gw)
 }
 
 
-void AdapterManager::initialize(void)
+void AdapterManager::initialize(char* gwName, bool aggregate, bool forwarder, bool qosM1)
 {
-    _aggregater->initialize();
-    _forwarders->initialize(_gateway);
-    _qosm1Proxy->initialize();
-}
+    if ( aggregate )
+    {
+    	_aggregater->initialize(gwName);
+    }
 
+    if ( qosM1 )
+    {
+    	_qosm1Proxy->initialize(gwName);
+    }
+
+    if ( forwarder )
+    {
+    	_forwarders->initialize(_gateway);
+    }
+}
 
 AdapterManager::~AdapterManager(void)
 {
@@ -91,22 +101,29 @@ bool AdapterManager::isAggregatedClient(Client* client)
 	}
 }
 
-Client* AdapterManager::getClient(Client& client)
+Client* AdapterManager::getClient(Client* client)
 {
-	bool secure = client.isSecureNetwork();
-	Client* newClient = &client;
-	if ( client.isQoSm1() )
+	bool secure = client->isSecureNetwork();
+	Client* newClient = client;
+
+	if ( client->isQoSm1() )
 	{
-		newClient = _qosm1Proxy->getAdapterClient(&client);
+		newClient = _qosm1Proxy->getAdapterClient(client);
 		_qosm1Proxy->resetPingTimer(secure);
 	}
-	else if ( client.isAggregated() )
-
+	else if ( client->isAggregated() )
+	{
+		newClient = _aggregater->getAdapterClient(client);
+		_aggregater->resetPingTimer(secure);
+	}
+	else if ( client->isQoSm1Proxy() )
+	{
+		_qosm1Proxy->resetPingTimer(secure);
+	}
+	else if ( client->isAggregater() )
 	{
-		newClient = _aggregater->getAdapterClient(&client);
 		_aggregater->resetPingTimer(secure);
 	}
-
 	return newClient;
 }
 
@@ -121,8 +138,8 @@ int AdapterManager::unicastToClient(Client* client, MQTTSNPacket* packet, Client
 		MQTTSNGWEncapsulatedPacket encap(packet);
 		WirelessNodeId* wnId = fwd->getWirelessNodeId(client);
 		encap.setWirelessNodeId(wnId);
-		WRITELOG(FORMAT_Y_W_G, currentDateTime(), encap.getName(), RIGHTARROW, fwd->getId(), encap.print(pbuf));
 		task->log(client, packet);
+		WRITELOG(FORMAT_Y_W_G, currentDateTime(), encap.getName(), RIGHTARROW, fwd->getId(), encap.print(pbuf));
 		rc = encap.unicast(_gateway->getSensorNetwork(),fwd->getSensorNetAddr());
 	}
 	else
@@ -167,22 +184,26 @@ bool AdapterManager::isAggregaterActive(void)
 	return _aggregater->isActive();
 }
 
-AggregateTopicElement* AdapterManager::createClientList(Topic* topic)
+/*
+AggregateTopicElement* AdapterManager::findTopic(Topic* topic)
 {
-	return _aggregater->createClientList(topic);
+	return _aggregater->findTopic(topic);
 }
 
-int AdapterManager::addAggregateTopic(Topic* topic, Client* client)
+AggregateTopicElement* AdapterManager::addAggregateTopic(Topic* topic, Client* client)
 {
 	return _aggregater->addAggregateTopic(topic, client);
 }
 
+
 void AdapterManager::removeAggregateTopic(Topic* topic, Client* client)
 {
-	 _aggregater->removeAggregateTopic(topic, client);
+	 //_aggregater->removeAggregateTopic(topic, client);
 }
 
 void AdapterManager::removeAggregateTopicList(Topics* topics, Client* client)
 {
-	 _aggregater->removeAggregateTopicList(topics, client);
+
 }
+*/
+

MQTTSNGateway/src/MQTTSNGWAdapterManager.h

@@ -40,21 +40,17 @@ class AdapterManager
 public:
 	AdapterManager(Gateway* gw);
     ~AdapterManager(void);
-    void initialize(void);
+    void initialize(char* gwName, bool aggregater, bool fowarder, bool qosM1);
     ForwarderList* getForwarderList(void);
     QoSm1Proxy* getQoSm1Proxy(void);
     Aggregater* getAggregater(void);
     void checkConnection(void);
 
     bool isAggregatedClient(Client* client);
-    Client* getClient(Client& client);
+    Client* getClient(Client* client);
     Client* convertClient(uint16_t msgId, uint16_t* clientMsgId);
     int unicastToClient(Client* client, MQTTSNPacket* packet, ClientSendTask* task);
     bool isAggregaterActive(void);
-    AggregateTopicElement* createClientList(Topic* topic);
-    int addAggregateTopic(Topic* topic, Client* client);
-    void removeAggregateTopic(Topic* topic, Client* client);
-    void removeAggregateTopicList(Topics* topics, Client* client);
 
 private:
     Gateway* _gateway {nullptr};

MQTTSNGateway/src/MQTTSNGWAggregateTopicTable.cpp

@@ -34,6 +34,11 @@ Client* ClientTopicElement::getClient(void)
 	return _client;
 }
 
+ClientTopicElement* ClientTopicElement::getNextClientElement(void)
+{
+	return _next;
+}
+
 /*=====================================
  Class AggregateTopicElement
  =====================================*/
@@ -44,6 +49,7 @@ AggregateTopicElement::AggregateTopicElement(void)
 
 AggregateTopicElement::AggregateTopicElement(Topic* topic, Client* client)
 {
+	_topic = topic;
 	ClientTopicElement* elm = new ClientTopicElement(client);
 	if ( elm != nullptr )
 	{
@@ -76,7 +82,9 @@ ClientTopicElement* AggregateTopicElement::add(Client* client)
 	{
 		return nullptr;
 	}
+
 	_mutex.lock();
+
 	if ( _head == nullptr )
 	{
 		_head = elm;
@@ -95,7 +103,7 @@ ClientTopicElement* AggregateTopicElement::add(Client* client)
 		else
 		{
 			delete elm;
-			elm = nullptr;
+			elm = p;
 		}
 	}
 	_mutex.unlock();
@@ -105,7 +113,7 @@ ClientTopicElement* AggregateTopicElement::add(Client* client)
 ClientTopicElement* AggregateTopicElement::find(Client* client)
 {
 	ClientTopicElement* p = _head;
-	while ( p )
+	while ( p != nullptr )
 	{
 		if ( p->_client == client)
 		{
@@ -116,16 +124,48 @@ ClientTopicElement* AggregateTopicElement::find(Client* client)
 	return p;
 }
 
-ClientTopicElement* AggregateTopicElement::getFirstElement(void)
+ClientTopicElement* AggregateTopicElement::getFirstClientTopicElement(void)
 {
 	return _head;
 }
 
-ClientTopicElement* AggregateTopicElement::getNextElement(ClientTopicElement* elm)
+ClientTopicElement* AggregateTopicElement::getNextClientTopicElement(ClientTopicElement* elmClient)
 {
-	return elm->_next;
+	return elmClient->_next;
 }
 
+void AggregateTopicElement::eraseClient(Client* client)
+{
+	_mutex.lock();
+
+	ClientTopicElement* p = find(client);
+	if ( p != nullptr )
+	{
+		if ( p->_prev == nullptr )    // head element
+		{
+			_head = p->_next;
+			if ( p->_next == nullptr )   // head & only one
+			{
+				_tail = nullptr;
+			}
+			else
+			{
+				p->_next->_prev = nullptr;  // head & midle
+			}
+		}
+		else if ( p->_next != nullptr )  // middle
+		{
+			p->_prev->_next = p->_next;
+		}
+		else    // tail
+		{
+			p->_prev->_next = nullptr;
+			_tail = p->_prev;
+		}
+		delete p;
+	}
+	_mutex.unlock();
+}
 
 /*=====================================
  Class AggregateTopicTable
@@ -143,19 +183,138 @@ AggregateTopicTable::~AggregateTopicTable()
 
 AggregateTopicElement* AggregateTopicTable::add(Topic* topic, Client* client)
 {
-	//ToDo: AggregateGW
-	return 0;
+	AggregateTopicElement* elm = nullptr;
+	_mutex.lock();
+	elm = getAggregateTopicElement(topic);
+	if ( elm != nullptr )
+	{
+		if ( elm->find(client) == nullptr )
+		{
+			elm->add(client);
+		}
+	}
+	else
+	{
+		Topic* newTopic = topic->duplicate();
+		elm = new AggregateTopicElement(newTopic, client);
+		if ( _head == nullptr )
+		{
+			_head = elm;
+			_tail = elm;
+		}
+		else
+		{
+			elm->_prev = _tail;
+			_tail->_next = elm;
+			_tail = elm;
+		}
+	}
+	_mutex.unlock();
+	return elm;
 }
 
-void AggregateTopicTable::remove(Topic* topic, Client* client)
+void AggregateTopicTable::erase(Topic* topic, Client* client)
 {
-	//ToDo: AggregateGW
+	AggregateTopicElement* elm = nullptr;
+
+	_mutex.lock();
+	elm = getAggregateTopicElement(topic);
+
+	if ( elm != nullptr )
+	{
+		elm->eraseClient(client);
+	}
+	if ( elm->_head == nullptr )
+	{
+		erase(elm);
+	}
+	_mutex.unlock();
+	return;
 }
 
-AggregateTopicElement* AggregateTopicTable::getClientList(Topic* client)
+void AggregateTopicTable::erase(AggregateTopicElement* elmTopic)
 {
-	// ToDo: AggregateGW
-	return 0;
+	if ( elmTopic != nullptr )
+	{
+		if ( elmTopic->_prev == nullptr )    // head element
+		{
+			_head = elmTopic->_next;
+			if ( elmTopic->_next == nullptr )   // head & only one
+			{
+				_tail = nullptr;
+			}
+			else
+			{
+				elmTopic->_next->_prev = nullptr;  // head & midle
+			}
+		}
+		else if ( elmTopic->_next != nullptr )  // middle
+		{
+			elmTopic->_prev->_next = elmTopic->_next;
+		}
+		else    // tail
+		{
+			elmTopic->_prev->_next = nullptr;
+			_tail = elmTopic->_prev;
+		}
+		delete elmTopic;
+	}
 }
 
+AggregateTopicElement* AggregateTopicTable::getAggregateTopicElement(Topic* topic)
+{
+	AggregateTopicElement* elm = _head;
 
+	while( elm != nullptr )
+	{
+		if ( elm->_topic->isMatch(topic->_topicName) )
+		{
+			break;
+		}
+		elm = elm->_next;
+	}
+	return elm;
+}
+
+ClientTopicElement* AggregateTopicTable::getClientElement(Topic* topic)
+{
+	AggregateTopicElement* elm = getAggregateTopicElement(topic);
+	if ( elm != nullptr )
+	{
+		return elm->_head;
+	}
+	else
+	{
+		return nullptr;
+	}
+}
+
+void AggregateTopicTable::print(void)
+{
+	AggregateTopicElement* elm = _head;
+
+	printf("Beginning of AggregateTopicTable\n");
+	while( elm != nullptr )
+	{
+		printf("%s\n", elm->_topic->getTopicName()->c_str());
+
+		ClientTopicElement* clElm = elm->getFirstClientTopicElement();
+		Client* client = clElm->getClient();
+
+		while ( client != nullptr )
+		{
+			printf("    %s\n", client->getClientId());
+			clElm = clElm->getNextClientElement();
+			if ( clElm != nullptr )
+			{
+				client = clElm->getClient();
+			}
+			else
+			{
+				client = nullptr;
+			}
+		}
+		elm = elm->_next;
+	}
+	printf("End of AggregateTopicTable\n");
+}

MQTTSNGateway/src/MQTTSNGWAggregateTopicTable.h

@@ -39,10 +39,16 @@ public:
 	~AggregateTopicTable();
 
 	AggregateTopicElement* add(Topic* topic, Client* client);
-	AggregateTopicElement* getClientList(Topic* client);
-	void remove(Topic* topic, Client* client);
+	AggregateTopicElement* getAggregateTopicElement(Topic* topic);
+	ClientTopicElement* getClientElement(Topic* topic);
+	void erase(Topic* topic, Client* client);
 	void clear(void);
+
+	void print(void);
+
 private:
+	void erase(AggregateTopicElement* elmTopic);
+	Mutex _mutex;
 	AggregateTopicElement* _head {nullptr};
 	AggregateTopicElement* _tail {nullptr};
 	int _cnt {0};
@@ -61,14 +67,16 @@ public:
     ~AggregateTopicElement(void);
 
     ClientTopicElement* add(Client* client);
-    ClientTopicElement* getFirstElement(void);
-    ClientTopicElement* getNextElement(ClientTopicElement* elm);
-    void erase(ClientTopicElement* elm);
+    ClientTopicElement* getFirstClientTopicElement(void);
+    ClientTopicElement* getNextClientTopicElement(ClientTopicElement* elmClient);
+    void eraseClient(Client* client);
     ClientTopicElement* find(Client* client);
 
 private:
     Mutex _mutex;
     Topic* _topic {nullptr};
+    AggregateTopicElement* _next {nullptr};
+    AggregateTopicElement* _prev {nullptr};
     ClientTopicElement* _head {nullptr};
     ClientTopicElement* _tail {nullptr};
 };
@@ -83,6 +91,8 @@ class ClientTopicElement
 public:
     ClientTopicElement(Client* client);
     ~ClientTopicElement(void);
+
+    ClientTopicElement* getNextClientElement(void);
     Client* getClient(void);
 
 private:

MQTTSNGateway/src/MQTTSNGWAggregater.cpp

@@ -36,22 +36,12 @@ Aggregater::~Aggregater(void)
 
 }
 
-void Aggregater::initialize(void)
+void Aggregater::initialize(char* gwName)
 {
-    char param[MQTTSNGW_PARAM_MAX];
-
-    if (_gateway->getParam("AggregatingGateway", param) == 0 )
-    {
-        if (!strcasecmp(param, "YES") )
-        {
-           /* Create Aggregated Clients */
-        	_gateway->getClientList()->setClientList(AGGREGATER_TYPE);
-
-        	string name = _gateway->getGWParams()->gatewayName;
-        	setup(name.c_str(), Atype_Aggregater);
-        	_isActive = true;
-        }
-    }
+	/* Create Aggregater Client */
+	string name = string(gwName) + string("_Aggregater");
+	setup(name.c_str(), Atype_Aggregater);
+	_isActive = true;
 
     //testMessageIdTable();
 
@@ -64,6 +54,7 @@ bool Aggregater::isActive(void)
 
 uint16_t Aggregater::msgId(void)
 {
+	// Only SecureClient generates msgId to avoid duplication of msgId. Client does not generate it.
 	return Adapter::getSecureClient()->getNextPacketId();
 }
 
@@ -93,26 +84,38 @@ uint16_t Aggregater::getMsgId(Client* client, uint16_t clientMsgId)
 	return _msgIdTable.getMsgId(client, clientMsgId);
 }
 
+AggregateTopicElement* Aggregater::addAggregateTopic(Topic* topic, Client* client)
+{
+	return _topicTable.add(topic, client);
+}
+
+
 void Aggregater::removeAggregateTopic(Topic* topic, Client* client)
 {
-      // ToDo: AggregateGW this method called when the client disconnect and erase it`s Topics. this method call */
+	_topicTable.erase(topic, client);
 }
 
-void Aggregater::removeAggregateTopicList(Topics* topics, Client* client)
+AggregateTopicElement* Aggregater::findTopic(Topic* topic)
 {
-      // ToDo: AggregateGW this method called when the client disconnect and erase it`s Topics. this method call */
+	return _topicTable.getAggregateTopicElement(topic);
 }
 
-int Aggregater::addAggregateTopic(Topic* topic, Client* client)
+ClientTopicElement* Aggregater::getClientElement(Topic* topic)
 {
-	// ToDo: AggregateGW  */
-	return 0;
+	AggregateTopicElement* elm = findTopic(topic);
+	if ( elm != nullptr )
+	{
+		return elm->getFirstClientTopicElement();
+	}
+	else
+	{
+		return nullptr;
+	}
 }
 
-AggregateTopicElement* Aggregater::createClientList(Topic* topic)
+void Aggregater::printAggregateTopicTable(void)
 {
-	// ToDo: AggregateGW  */
-	return 0;
+	_topicTable.print();
 }
 
 bool Aggregater::testMessageIdTable(void)

MQTTSNGateway/src/MQTTSNGWAggregater.h

@@ -20,6 +20,7 @@
 #include "MQTTSNGWAdapter.h"
 #include "MQTTSNGWMessageIdTable.h"
 #include "MQTTSNGWAggregateTopicTable.h"
+
 namespace MQTTSNGW
 {
 class Gateway;
@@ -40,7 +41,7 @@ public:
     Aggregater(Gateway* gw);
     ~Aggregater(void);
 
-    void initialize(void);
+    void initialize(char* gwName);
 
     const char* getClientId(SensorNetAddress* addr);
 	Client* getClient(SensorNetAddress* addr);
@@ -48,13 +49,18 @@ public:
 	uint16_t addMessageIdTable(Client* client, uint16_t msgId);
 	uint16_t getMsgId(Client* client, uint16_t clientMsgId);
 
+	ClientTopicElement* getClientElement(Topic* topic);
+	ClientTopicElement* getNextClientElement(ClientTopicElement* clientElement);
+	Client* getClient(ClientTopicElement* clientElement);
+
+	AggregateTopicElement* findTopic(Topic* topic);
+	AggregateTopicElement* addAggregateTopic(Topic* topic, Client* client);
 
-	AggregateTopicElement* createClientList(Topic* topic);
-	int addAggregateTopic(Topic* topic, Client* client);
 	void removeAggregateTopic(Topic* topic, Client* client);
-	void removeAggregateTopicList(Topics* topics, Client* client);
+	void removeAggregateAllTopic(Client* client);
 	bool isActive(void);
 
+	void printAggregateTopicTable(void);
 	bool testMessageIdTable(void);
 
 private:

MQTTSNGateway/src/MQTTSNGWBrokerRecvTask.cpp

@@ -64,7 +64,7 @@ void BrokerRecvTask::run(void)
 		_light->blueLight(false);
 		if (CHK_SIGINT)
 		{
-			WRITELOG("%s BrokerRecvTask   stopped.\n", currentDateTime());
+			WRITELOG("\n%s BrokerRecvTask   stopped.", currentDateTime());
 			return;
 		}
 		timeout.tv_sec = 0;
@@ -161,7 +161,7 @@ void BrokerRecvTask::run(void)
 
 								delete packet;
 
-								if ( (rc == -1 || rc == -2) && client->isActive() )
+								if ( (rc == -1 || rc == -2) && ( client->isActive()  || client->isSleep() || client->isAwake() ))
 								{
 									/* disconnect the client */
 									packet = new MQTTGWPacket();

MQTTSNGateway/src/MQTTSNGWBrokerSendTask.cpp

@@ -41,7 +41,7 @@ BrokerSendTask::BrokerSendTask(Gateway* gateway)
 
 BrokerSendTask::~BrokerSendTask()
 {
-
+//	WRITELOG("BrokerSendTask is deleted normally.\r\n");
 }
 
 /**
@@ -70,7 +70,7 @@ void BrokerSendTask::run()
 
 		if ( ev->getEventType() == EtStop )
 		{
-			WRITELOG("%s BrokerSendTask   stopped.\n", currentDateTime());
+			WRITELOG("\n%s BrokerSendTask   stopped.", currentDateTime());
 			delete ev;
 			return;
 		}
@@ -81,7 +81,7 @@ void BrokerSendTask::run()
 			packet = ev->getMQTTGWPacket();
 
 			/* Check Client is managed by Adapters */
-			client = adpMgr->getClient(*client);
+			client = adpMgr->getClient(client);
 
 			if ( packet->getType() == CONNECT && client->getNetwork()->isValid() )
 			{
@@ -121,13 +121,21 @@ void BrokerSendTask::run()
 				{
 					client->connectSended();
 				}
+				else if ( packet->getType() == DISCONNECT )
+				{
+					client->getNetwork()->close();
+					client->disconnected();
+				}
 				log(client, packet);
 			}
 			else
 			{
 				WRITELOG("%s BrokerSendTask: %s can't send a packet to the broker. errno=%d %s %s\n",
 						ERRMSG_HEADER, client->getClientId(), rc == -1 ? errno : 0, strerror(errno), ERRMSG_FOOTER);
-				client->getNetwork()->close();
+				if ( errno != EBADF )
+				{
+					client->getNetwork()->close();
+				}
 
 				/* Disconnect the client */
 				packet = new MQTTGWPacket();

MQTTSNGateway/src/MQTTSNGWClient.cpp

@@ -572,7 +572,7 @@ void Client::resetPingRequest(void)
     _holdPingRequest = false;
 }
 
-bool Client::isHoldPringReqest(void)
+bool Client::isHoldPingReqest(void)
 {
     return _holdPingRequest;
 }

MQTTSNGateway/src/MQTTSNGWClient.h

@@ -252,7 +252,7 @@ public:
 
     void holdPingRequest(void);
     void resetPingRequest(void);
-    bool isHoldPringReqest(void);
+    bool isHoldPingReqest(void);
 
     Client* getNextClient(void);
 

MQTTSNGateway/src/MQTTSNGWClientList.cpp

@@ -61,52 +61,27 @@ void ClientList::initialize(bool aggregate)
 		setClientList(type);
         _authorize = true;
     }
+
+    if ( theGateway->getGWParams()->predefinedTopic )
+    {
+    	setPredefinedTopics(aggregate);
+    }
 }
 
 void ClientList::setClientList(int type)
 {
-	char param[MQTTSNGW_PARAM_MAX];
-	string fileName;
-	GatewayParams* params = theGateway->getGWParams();
-	if (theGateway->getParam("ClientsList", param) == 0)
+	if (!createList(theGateway->getGWParams()->clientListName, type))
 	{
-		fileName = string(param);
-	}
-	else
-	{
-		fileName = params->configDir + string(CLIENT_LIST);
-	}
-
-	if (!createList(fileName.c_str(), type))
-	{
-		throw Exception("ClientList::initialize(): No client list defined by the configuration.");
-	}
-
-	if ( params->clientListName == nullptr )
-	{
-		params->clientListName = strdup(fileName.c_str());
+		throw Exception("ClientList::setClientList No client list defined by config file.");
 	}
 }
 
 void ClientList::setPredefinedTopics(bool aggrecate)
 {
-	char param[MQTTSNGW_PARAM_MAX];
-
-	string fileName;
-	GatewayParams* params = theGateway->getGWParams();
-
-	if (theGateway->getParam("PredefinedTopicList", param) == 0)
+	if ( !readPredefinedList(theGateway->getGWParams()->predefinedTopicFileName, aggrecate) )
 	{
-		fileName = string(param);
-	}
-	else
-	{
-		fileName = params->configDir + string(PREDEFINEDTOPIC_FILE);
-	}
+		throw Exception("ClientList::setPredefinedTopics No predefindTopi list defined by config file.");
 
-	if ( readPredefinedList(fileName.c_str(), aggrecate) )
-	{
-		params->predefinedTopicFileName = strdup(fileName.c_str());
 	}
 }
 
@@ -118,18 +93,18 @@ void ClientList::setPredefinedTopics(bool aggrecate)
  * File format is:
  *     Lines bigning with # are comment line.
  *     ClientId, SensorNetAddress, "unstableLine", "secureConnection"
- *     in case of UDP, SensorNetAddress format is portNo@IPAddress.
+ *     in case of UDP, SensorNetAddress format is IPAddress:portNo.
  *     if the SensorNetwork is not stable, write unstableLine.
  *     if BrokerConnection is SSL, write secureConnection.
  *     if the client send PUBLISH QoS-1, QoS-1 is required.
  *
  * Ex:
  *     #Client List
- *     ClientId1,11200@192.168.10.10
- *     ClientID2,35000@192.168.50.200,unstableLine
- *     ClientID3,40000@192.168.200.50,secureConnection
- *     ClientID4,41000@192.168.200.51,unstableLine,secureConnection
- *      ClientID5,41000@192.168.200.51,unstableLine,secureConnection,QoS-1
+ *     ClientId1,192.168.10.10:11200
+ *     ClientID2,192.168.50.200:35000,unstableLine
+ *     ClientID3,192.168.200.50:40000,secureConnection
+ *     ClientID4,192.168.200.51:41000,unstableLine,secureConnection
+ *     ClientID5,192.168.200.51:41000,unstableLine,secureConnection,QoS-1
  */
 
 bool ClientList::createList(const char* fileName, int type)
@@ -141,7 +116,7 @@ bool ClientList::createList(const char* fileName, int type)
     bool stable;
     bool qos_1;
     bool forwarder;
-    bool rc = true;
+    bool rc = false;
     SensorNetAddress netAddr;
     MQTTSNString clientId = MQTTSNString_initializer;
 
@@ -194,6 +169,7 @@ bool ClientList::createList(const char* fileName, int type)
             free(clientId.cstring);
         }
         fclose(fp);
+        rc = true;
     }
     return rc;
 }
@@ -380,7 +356,7 @@ Client* ClientList::createClient(SensorNetAddress* addr, MQTTSNString* clientId,
     }
     else
     {
-        MQTTSNString  dummyId MQTTSNString_initializer;;
+        MQTTSNString  dummyId MQTTSNString_initializer;
         dummyId.cstring = strdup("");
         client->setClientId(dummyId);
          free(dummyId.cstring);
@@ -416,10 +392,16 @@ Client* ClientList::createClient(SensorNetAddress* addr, MQTTSNString* clientId,
 
 Client* ClientList::createPredefinedTopic( MQTTSNString* clientId, string topicName, uint16_t topicId, bool aggregate)
 {
+	if ( topicId == 0 )
+	{
+		WRITELOG("Invalid TopicId. Predefined Topic %s,  TopicId is 0. \n", topicName.c_str());
+		return nullptr;
+	}
+
 	if ( strcmp(clientId->cstring, common_topic) == 0 )
 	{
 		theGateway->getTopics()->add((const char*)topicName.c_str(), topicId);
-		return 0;
+		return nullptr;
 	}
 	else
 	{
@@ -427,7 +409,7 @@ Client* ClientList::createPredefinedTopic( MQTTSNString* clientId, string topicN
 
 		if ( _authorize && client == nullptr )
 		{
-			return 0;
+			return nullptr;
 		}
 
 		/*  anonimous clients */

MQTTSNGateway/src/MQTTSNGWClientRecvTask.cpp

@@ -60,7 +60,7 @@ void ClientRecvTask::run()
 	Event* ev = nullptr;
 	AdapterManager* adpMgr = _gateway->getAdapterManager();
 	QoSm1Proxy* qosm1Proxy = adpMgr->getQoSm1Proxy();
-	bool isAggrActive = adpMgr->isAggregaterActive();
+	int  clientType = adpMgr->isAggregaterActive() ? AGGREGATER_TYPE : TRANSPEARENT_TYPE;
 	ClientList* clientList = _gateway->getClientList();
 	EventQue* packetEventQue = _gateway->getPacketEventQue();
 
@@ -77,7 +77,7 @@ void ClientRecvTask::run()
 
 		if (CHK_SIGINT)
 		{
-			WRITELOG("%s ClientRecvTask   stopped.\n", currentDateTime());
+			WRITELOG("\n%s ClientRecvTask   stopped.", currentDateTime());
 			delete packet;
 			return;
 		}
@@ -133,11 +133,12 @@ void ClientRecvTask::run()
 			{
 				 const char* clientName = qosm1Proxy->getClientId(senderAddr);
 
-				if ( clientName )
+				if ( clientName != nullptr )
 				{
+					client = qosm1Proxy->getClient();
+
 					if ( !packet->isQoSMinusPUBLISH() )
 					{
-						client = qosm1Proxy->getClient();
 						log(clientName, packet);
 						WRITELOG("%s %s  %s can send only PUBLISH with QoS-1.%s\n", ERRMSG_HEADER, clientName, senderAddr->sprint(buf), ERRMSG_FOOTER);
 						delete packet;
@@ -145,11 +146,14 @@ void ClientRecvTask::run()
 					}
 				}
 			}
+
+			if ( client == nullptr )
+			{
+				client = _gateway->getClientList()->getClient(senderAddr);
+			}
 		}
 
-		client = _gateway->getClientList()->getClient(senderAddr);
-
-		if ( client )
+		if ( client != nullptr )
 		{
 			/* write log and post Event */
 			log(client, packet, 0);
@@ -174,33 +178,36 @@ void ClientRecvTask::run()
 
 				client = clientList->getClient(&data.clientID);
 
-				if ( fwd )
+				if ( fwd != nullptr )
 				{
 				    if ( client == nullptr )
 				    {
 				        /* create a new client */
-				        client = clientList->createClient(0, &data.clientID, isAggrActive);
+				        client = clientList->createClient(0, &data.clientID, clientType);
 				    }
-				    /* Add to af forwarded client list of forwarder. */
+				    /* Add to a forwarded client list of forwarder. */
                     fwd->addClient(client, &nodeId);
 				}
 				else
 				{
                     if ( client )
                     {
-                        /* Client exists. Set SensorNet Address of it. */
-                        client->setClientAddress(senderAddr);
+                        /* Authentication is not required */
+                    	if ( _gateway->getGWParams()->clientAuthentication == false)
+                    	{
+                    		client->setClientAddress(senderAddr);
+                    	}
                     }
                     else
                     {
                         /* create a new client */
-                        client = clientList->createClient(senderAddr, &data.clientID, isAggrActive);
+                        client = clientList->createClient(senderAddr, &data.clientID, clientType);
                     }
 				}
 
 				log(client, packet, &data.clientID);
 
-				if (!client)
+				if ( client == nullptr )
 				{
 	                WRITELOG("%s Client(%s) was rejected. CONNECT message has been discarded.%s\n", ERRMSG_HEADER, senderAddr->sprint(buf), ERRMSG_FOOTER);
 					delete packet;
@@ -217,11 +224,11 @@ void ClientRecvTask::run()
 				log(client, packet, 0);
 				if ( packet->getType() == MQTTSN_ENCAPSULATED )
 				{
-					WRITELOG("%s Forwarder(%s) is not declared by ClientList file. message has been discarded.%s\n", ERRMSG_HEADER, _sensorNetwork->getSenderAddress()->sprint(buf), ERRMSG_FOOTER);
+					WRITELOG("%s MQTTSNGWClientRecvTask  Forwarder(%s) is not declared by ClientList file. message has been discarded.%s\n", ERRMSG_HEADER, _sensorNetwork->getSenderAddress()->sprint(buf), ERRMSG_FOOTER);
 				}
 				else
 				{
-					WRITELOG("%s Client(%s) is not connecting. message has been discarded.%s\n", ERRMSG_HEADER, senderAddr->sprint(buf), ERRMSG_FOOTER);
+					WRITELOG("%s MQTTSNGWClientRecvTask  Client(%s) is not connecting. message has been discarded.%s\n", ERRMSG_HEADER, senderAddr->sprint(buf), ERRMSG_FOOTER);
 				}
 				delete packet;
 			}

MQTTSNGateway/src/MQTTSNGWClientSendTask.cpp

@@ -35,7 +35,7 @@ ClientSendTask::ClientSendTask(Gateway* gateway)
 
 ClientSendTask::~ClientSendTask()
 {
-
+//	WRITELOG("ClientSendTask is deleted normally.\r\n");
 }
 
 void ClientSendTask::run()
@@ -49,35 +49,44 @@ void ClientSendTask::run()
 	{
 		Event* ev = _gateway->getClientSendQue()->wait();
 
-		if (ev->getEventType() == EtStop)
+		if (ev->getEventType() == EtStop || _gateway->IsStopping() )
 		{
-			WRITELOG("%s ClientSendTask   stopped.\n", currentDateTime());
+			WRITELOG("\n%s ClientSendTask   stopped.", currentDateTime());
 			delete ev;
 			break;
 		}
-		if (ev->getEventType() == EtClientSend)
-		{
-			client = ev->getClient();
-			packet = ev->getMQTTSNPacket();
-			rc = adpMgr->unicastToClient(client, packet, this);
-		}
-		else if (ev->getEventType() == EtBroadcast)
-		{
-			packet = ev->getMQTTSNPacket();
-			log(client, packet);
-			rc = packet->broadcast(_sensorNetwork);
-		}
-		else if (ev->getEventType() == EtSensornetSend)
-		{
-			packet = ev->getMQTTSNPacket();
-			log(client, packet);
-			rc = packet->unicast(_sensorNetwork, ev->getSensorNetAddress());
-		}
 
-		if ( rc < 0 )
+		if (ev->getEventType() == EtBroadcast)
 		{
-			WRITELOG("%s ClientSendTask can't send a packet to the client %s. Error=%d%s\n",
-				ERRMSG_HEADER, (client ? (const char*)client->getClientId() : UNKNOWNCL ), errno, ERRMSG_FOOTER);
+			packet = ev->getMQTTSNPacket();
+			log(client, packet);
+
+			if ( packet->broadcast(_sensorNetwork) < 0 )
+			{
+				WRITELOG("%s ClientSendTask can't multicast a packet Error=%d%s\n",
+					ERRMSG_HEADER, errno, ERRMSG_FOOTER);
+			}
+		}
+		else
+		{
+			if (ev->getEventType() == EtClientSend)
+			{
+				client = ev->getClient();
+				packet = ev->getMQTTSNPacket();
+				rc = adpMgr->unicastToClient(client, packet, this);
+			}
+			else if (ev->getEventType() == EtSensornetSend)
+			{
+				packet = ev->getMQTTSNPacket();
+				log(client, packet);
+				rc = packet->unicast(_sensorNetwork, ev->getSensorNetAddress());
+			}
+
+			if ( rc < 0 )
+			{
+				WRITELOG("%s ClientSendTask can't send a packet to the client %s. Error=%d%s\n",
+					ERRMSG_HEADER, (client ? (const char*)client->getClientId() : UNKNOWNCL ), errno, ERRMSG_FOOTER);
+			}
 		}
 		delete ev;
 	}

MQTTSNGateway/src/MQTTSNGWDefines.h

@@ -42,13 +42,14 @@ namespace MQTTSNGW
 #define MAX_INFLIGHTMESSAGES         (10)  // Number of inflight messages
 #define MAX_MESSAGEID_TABLE_SIZE    (500)  // Number of MessageIdTable size
 #define MAX_SAVED_PUBLISH            (20)  // Max number of PUBLISH message for Asleep state
-#define MAX_TOPIC_PAR_CLIENT     (50)    // Max Topic count for a client. it should be less than 256
+#define MAX_TOPIC_PAR_CLIENT         (50)  // Max Topic count for a client. it should be less than 256
 #define MQTTSNGW_MAX_PACKET_SIZE   (1024)  // Max Packet size  (5+2+TopicLen+PayloadLen + Foward Encapsulation)
 #define SIZE_OF_LOG_PACKET          (500)  // Length of the packet log in bytes
 
-#define QOSM1_PROXY_KEEPALIVE_DURATION   900       // Secs
-#define QOSM1_PROXY_RESPONSE_DURATION     10       // Secs
-#define QOSM1_PROXY_MAX_RETRY_CNT        3
+#define PROXY_KEEPALIVE_DURATION   (900)   // Seconds
+#define PROXY_RESPONSE_DURATION     (10)   // Seconds
+#define PROXY_MAX_RETRY_CNT          (3)
+
 /*=================================
  *    Data Type
  ==================================*/

MQTTSNGateway/src/MQTTSNGWForwarder.cpp

@@ -47,16 +47,8 @@ ForwarderList::~ForwarderList()
 
 void ForwarderList::initialize(Gateway* gw)
 {
-    char param[MQTTSNGW_PARAM_MAX];
-    string fileName;
-
-    if (gw->getParam("Forwarder", param) == 0 )
-    {
-        if (!strcasecmp(param, "YES") )
-        {
-        	gw->getClientList()->setClientList(FORWARDER_TYPE);
-        }
-    }
+	/* Create Fowarders from clients.conf */
+	gw->getClientList()->setClientList(FORWARDER_TYPE);
 }
 
 
@@ -151,6 +143,7 @@ void Forwarder::addClient(Client* client, WirelessNodeId* id)
             if ( p->_client == client )
             {
                 client->setForwarder(this);
+                p->setWirelessNodeId(id);
                 return;
             }
             prev = p;

MQTTSNGateway/src/MQTTSNGWPacketHandleTask.cpp

@@ -113,7 +113,7 @@ void PacketHandleTask::run()
 
 		if (ev->getEventType() == EtStop)
 		{
-			WRITELOG("%s PacketHandleTask stopped.\n", currentDateTime());
+			WRITELOG("\n%s PacketHandleTask stopped.", currentDateTime());
 			delete ev;
 			return;
 		}
@@ -148,7 +148,7 @@ void PacketHandleTask::run()
 
 			if ( adpMgr->isAggregatedClient(client) )
 			{
-				aggregatePacketHandler(client, snPacket);
+				aggregatePacketHandler(client, snPacket);     // client is converted to Aggregater by BrokerSendTask
 			}
 			else
 			{
@@ -198,6 +198,9 @@ void PacketHandleTask::aggregatePacketHandler(Client*client, MQTTSNPacket* packe
 	case MQTTSN_DISCONNECT:
 		_mqttsnAggrConnection->handleDisconnect(client, packet);
 		break;
+	case MQTTSN_WILLTOPICUPD:
+		_mqttsnConnection->handleWilltopicupd(client, packet);
+		break;
 	case MQTTSN_WILLMSGUPD:
 		_mqttsnConnection->handleWillmsgupd(client, packet);
 		break;
@@ -289,6 +292,9 @@ void PacketHandleTask::transparentPacketHandler(Client*client, MQTTSNPacket* pac
 	case MQTTSN_DISCONNECT:
 		_mqttsnConnection->handleDisconnect(client, packet);
 		break;
+	case MQTTSN_WILLTOPICUPD:
+		_mqttsnConnection->handleWilltopicupd(client, packet);
+		break;
 	case MQTTSN_WILLMSGUPD:
 		_mqttsnConnection->handleWillmsgupd(client, packet);
 		break;
@@ -359,6 +365,9 @@ void PacketHandleTask::transparentPacketHandler(Client*client, MQTTGWPacket* pac
 	case UNSUBACK:
 		_mqttSubscribe->handleUnsuback(client, packet);
 		break;
+	case DISCONNECT:
+		client->disconnected();    // Just change Client's status to "Disconnected"
+		break;
 	default:
 		break;
 	}

MQTTSNGateway/src/MQTTSNGWProcess.cpp

@@ -103,7 +103,7 @@ void Process::initialize(int argc, char** argv)
 			}
 		}
 	}
-	_rbsem = new Semaphore(MQTTSNGW_RB_SEMAPHOR_NAME, 0);
+	_rbsem = new NamedSemaphore(MQTTSNGW_RB_SEMAPHOR_NAME, 0);
 	_rb = new RingBuffer(_configDir.c_str());
 
 	if (getParam("ShearedMemory", param) == 0)
@@ -306,7 +306,7 @@ void MultiTaskProcess::waitStop(void)
 	}
 }
 
-void MultiTaskProcess::threadStoped(void)
+void MultiTaskProcess::threadStopped(void)
 {
 	_mutex.lock();
 	_stopCount++;

MQTTSNGateway/src/MQTTSNGWProcess.h

@@ -66,7 +66,7 @@ private:
 	string  _configDir;
 	string  _configFile;
 	RingBuffer* _rb;
-	Semaphore*  _rbsem;
+	NamedSemaphore*  _rbsem;
 	Mutex _mt;
 	int  _log;
 	char _rbdata[PROCESS_LOG_BUFFER_SIZE + 1];
@@ -84,7 +84,7 @@ public:
 	int getParam(const char* parameter, char* value);
 	void run(void);
 	void waitStop(void);
-	void threadStoped(void);
+	void threadStopped(void);
 	void attach(Thread* thread);
 
 private:

MQTTSNGateway/src/MQTTSNGWPublishHandler.cpp

@@ -94,6 +94,12 @@ MQTTGWPacket* MQTTSNPublishHandler::handlePublish(Client* client, MQTTSNPacket*
 	        return nullptr;
 	    }
 
+	    if ( ( qos == 0 || qos == 3 ) && msgId > 0 )
+	    {
+	        WRITELOG("%s Invalid MsgId.%s %s\n", ERRMSG_HEADER, client->getClientId(), ERRMSG_FOOTER);
+	        return nullptr;
+	    }
+
 		if( !topic && msgId && qos > 0 && qos < 3 )
 		{
 			/* Reply PubAck with INVALID_TOPIC_ID to the client */
@@ -233,7 +239,7 @@ void MQTTSNPublishHandler::handleRegAck( Client* client, MQTTSNPacket* packet)
             ev->setClientSendEvent(client, regAck);
             _gateway->getClientSendQue()->post(ev);
         }
-        if (client->isHoldPringReqest() && client->getWaitREGACKPacketList()->getCount() == 0 )
+        if (client->isHoldPingReqest() && client->getWaitREGACKPacketList()->getCount() == 0 )
         {
             /* send PINGREQ to the broker */
            client->resetPingRequest();

MQTTSNGateway/src/MQTTSNGWQoSm1Proxy.cpp

@@ -38,28 +38,20 @@ QoSm1Proxy::~QoSm1Proxy(void)
 }
 
 
-void QoSm1Proxy::initialize(void)
+void QoSm1Proxy::initialize(char* gwName)
 {
-    char param[MQTTSNGW_PARAM_MAX];
-
     if ( _gateway->hasSecureConnection() )
     {
 		_isSecure = true;
     }
 
-    if (_gateway->getParam("QoS-1", param) == 0 )
-    {
-        if (strcasecmp(param, "YES") == 0 )
-        {
-        	/*  Create QoS-1 Clients */
-        	_gateway->getClientList()->setClientList(QOSM1PROXY_TYPE);
+	/*  Create QoS-1 Clients from clients.conf */
+	_gateway->getClientList()->setClientList(QOSM1PROXY_TYPE);
 
-            /* initialize Adapter */
-			string name = string(_gateway->getGWParams()->gatewayName) + "QoS-1";
-            setup(name.c_str(), Atype_QoSm1Proxy);
-           _isActive = true;
-        }
-    }
+	/* Create a client for  QoS-1 proxy */
+	string name = string(gwName) + string("_QoS-1");
+	setup(name.c_str(), Atype_QoSm1Proxy);
+   _isActive = true;
 }
 
 

MQTTSNGateway/src/MQTTSNGWQoSm1Proxy.h

@@ -35,7 +35,7 @@ public:
 	QoSm1Proxy(Gateway* gw);
     ~QoSm1Proxy(void);
 
-    void initialize(void);
+    void initialize(char* GWnAME);
     bool isActive(void);
 
 private:

MQTTSNGateway/src/MQTTSNGWSubscribeHandler.cpp

@@ -100,7 +100,8 @@ MQTTGWPacket* MQTTSNSubscribeHandler::handleSubscribe(Client* client, MQTTSNPack
         topicstr[0] = topicFilter.data.short_name[0];
         topicstr[1] = topicFilter.data.short_name[1];
         topicstr[2] = 0;
-        topicId = 0;
+        topicId = topicFilter.data.short_name[0] << 8;
+        topicId |= topicFilter.data.short_name[1];
         subscribe = new MQTTGWPacket();
         subscribe->setSUBSCRIBE(topicstr, (uint8_t)qos, (uint16_t)msgId);
     }
@@ -203,11 +204,6 @@ void MQTTSNSubscribeHandler::handleAggregateSubscribe(Client* client, MQTTSNPack
 
 	if ( subscribe != nullptr )
 	{
-		UTF8String str = subscribe->getTopic();
-		string* topicName = new string(str.data, str.len);
-		Topic topic = Topic(topicName, MQTTSN_TOPIC_TYPE_NORMAL);
-		 _gateway->getAdapterManager()->addAggregateTopic(&topic, client);
-
 		int msgId = 0;
 		if ( packet->isDuplicate() )
 		{
@@ -223,7 +219,13 @@ void MQTTSNSubscribeHandler::handleAggregateSubscribe(Client* client, MQTTSNPack
 			WRITELOG("%s MQTTSNSubscribeHandler can't create MessageIdTableElement  %s%s\n", ERRMSG_HEADER, client->getClientId(), ERRMSG_FOOTER);
 			return;
 		}
-WRITELOG("msgId=%d\n",msgId);
+
+		UTF8String str = subscribe->getTopic();
+		string* topicName = new string(str.data, str.len);    // topicName is delete by topic
+		Topic topic = Topic(topicName, MQTTSN_TOPIC_TYPE_NORMAL);
+
+		_gateway->getAdapterManager()->getAggregater()->addAggregateTopic(&topic, client);
+
 		subscribe->setMsgId(msgId);
 		Event* ev = new Event();
 		ev->setBrokerSendEvent(client, subscribe);
@@ -236,11 +238,6 @@ void MQTTSNSubscribeHandler::handleAggregateUnsubscribe(Client* client, MQTTSNPa
 	MQTTGWPacket* unsubscribe = handleUnsubscribe(client, packet);
 	if ( unsubscribe != nullptr )
 	{
-		UTF8String str = unsubscribe->getTopic();
-		string* topicName = new string(str.data, str.len);
-		Topic topic = Topic(topicName, MQTTSN_TOPIC_TYPE_NORMAL);
-		_gateway->getAdapterManager()->removeAggregateTopic(&topic, client);
-
 		int msgId = 0;
 		if ( packet->isDuplicate() )
 		{
@@ -256,6 +253,12 @@ void MQTTSNSubscribeHandler::handleAggregateUnsubscribe(Client* client, MQTTSNPa
 			WRITELOG("%s MQTTSNUnsubscribeHandler can't create MessageIdTableElement  %s%s\n", ERRMSG_HEADER, client->getClientId(), ERRMSG_FOOTER);
 			return;
 		}
+
+		UTF8String str = unsubscribe->getTopic();
+		string* topicName = new string(str.data, str.len);     // topicName is delete by topic
+		Topic topic = Topic(topicName, MQTTSN_TOPIC_TYPE_NORMAL);
+		_gateway->getAdapterManager()->getAggregater()->removeAggregateTopic(&topic, client);
+
 		unsubscribe->setMsgId(msgId);
 		Event* ev = new Event();
 		ev->setBrokerSendEvent(client, unsubscribe);

MQTTSNGateway/src/MQTTSNGWTopic.cpp

@@ -63,6 +63,15 @@ MQTTSN_topicTypes Topic::getType(void)
     return _type;
 }
 
+Topic* Topic::duplicate(void)
+{
+	Topic* newTopic = new Topic();
+	newTopic->_type = _type;
+	newTopic->_topicId = _topicId;
+	newTopic->_topicName = new string(_topicName->c_str());
+	return newTopic;
+}
+
 bool Topic::isMatch(string* topicName)
 {
 	string::size_type tlen = _topicName->size();
@@ -354,6 +363,16 @@ void Topics::eraseNormal(void)
     }
 }
 
+Topic* Topics::getFirstTopic(void)
+{
+	return _first;
+}
+
+Topic* Topics::getNextTopic(Topic* topic)
+{
+	return topic->_next;
+}
+
 void Topics::print(void)
 {
     Topic* topic = _first;

MQTTSNGateway/src/MQTTSNGWTopic.h

@@ -31,6 +31,7 @@ namespace MQTTSNGW
 class Topic
 {
     friend class Topics;
+    friend class AggregateTopicTable;
 public:
     Topic();
     Topic(string* topic, MQTTSN_topicTypes type);
@@ -39,7 +40,9 @@ public:
     uint16_t getTopicId(void);
     MQTTSN_topicTypes getType(void);
     bool isMatch(string* topicName);
+    Topic* duplicate(void);
     void print(void);
+
 private:
     MQTTSN_topicTypes _type;
     uint16_t _topicId;
@@ -59,6 +62,8 @@ public:
     Topic* add(const char* topicName, uint16_t id = 0);
     Topic* getTopicByName(const MQTTSN_topicid* topic);
     Topic* getTopicById(const MQTTSN_topicid* topicid);
+    Topic* getFirstTopic(void);
+    Topic* getNextTopic(Topic* topic);
     Topic* match(const MQTTSN_topicid* topicid);
     void eraseNormal(void);
     uint16_t getNextTopicId();

MQTTSNGateway/src/MQTTSNGWVersion.h

@@ -17,6 +17,6 @@
 #ifndef MQTTSNGWVERSION_H_IN_
 #define MQTTSNGWVERSION_H_IN_
 
-#define PAHO_GATEWAY_VERSION "1.3.1"
+#define PAHO_GATEWAY_VERSION "1.4.0"
 
 #endif /* MQTTSNGWVERSION_H_IN_ */

MQTTSNGateway/src/MQTTSNGateway.cpp

@@ -38,6 +38,7 @@ Gateway::Gateway(void)
     _clientList = new ClientList();
     _adapterManager = new AdapterManager(this);
     _topics = new Topics();
+    _stopFlg = false;
 }
 
 Gateway::~Gateway()
@@ -86,6 +87,10 @@ Gateway::~Gateway()
 	{
 		free(_params.clientListName);
 	}
+	if ( _params.predefinedTopicFileName )
+	{
+		free( _params.predefinedTopicFileName);
+	}
 	if ( _params.configName )
 	{
 		free(_params.configName);
@@ -109,6 +114,7 @@ Gateway::~Gateway()
 	{
 		delete _topics;
 	}
+//    WRITELOG("Gateway        is deleted normally.\r\n");
 }
 
 int Gateway::getParam(const char* parameter, char* value)
@@ -116,6 +122,16 @@ int Gateway::getParam(const char* parameter, char* value)
     return MultiTaskProcess::getParam(parameter, value);
 }
 
+char* Gateway::getClientListFileName(void)
+{
+	return _params.clientListName;
+}
+
+char* Gateway::getPredefinedTopicFileName(void)
+{
+	return _params.predefinedTopicFileName;
+}
+
 void Gateway::initialize(int argc, char** argv)
 {
 	char param[MQTTSNGW_PARAM_MAX];
@@ -215,14 +231,53 @@ void Gateway::initialize(int argc, char** argv)
 		}
 	}
 
-	/*  ClientList and Adapters  Initialize  */
-	_adapterManager->initialize();
+	if (getParam("ClientsList", param) == 0)
+	{
+		_params.clientListName = strdup(param);
+	}
 
-	bool aggregate = _adapterManager->isAggregaterActive();
-	_clientList->initialize(aggregate);
+	if (getParam("PredefinedTopic", param) == 0)
+	{
+		if ( !strcasecmp(param, "YES") )
+		{
+			_params.predefinedTopic = true;
+			if (getParam("PredefinedTopicList", param) == 0)
+			{
+				_params.predefinedTopicFileName = strdup(param);
+			}
+		}
+	}
 
-	/*  Setup predefined topics  */
-	_clientList->setPredefinedTopics(aggregate);
+	if (getParam("AggregatingGateway", param) == 0)
+	{
+		if ( !strcasecmp(param, "YES") )
+		{
+			_params.aggregatingGw = true;
+		}
+	}
+
+	if (getParam("Forwarder", param) == 0)
+	{
+		if ( !strcasecmp(param, "YES") )
+		{
+			_params.forwarder = true;
+		}
+	}
+
+	if (getParam("QoS-1", param) == 0)
+	{
+		if ( !strcasecmp(param, "YES") )
+		{
+			_params.qosMinus1 = true;
+		}
+	}
+
+
+		/*  Initialize adapters */
+		_adapterManager->initialize( _params.gatewayName, _params.aggregatingGw, _params.forwarder, _params.qosMinus1);
+
+		/*  Setup ClientList and Predefined topics  */
+		_clientList->initialize(_params.aggregatingGw);
 }
 
 void Gateway::run(void)
@@ -256,10 +311,13 @@ void Gateway::run(void)
 	WRITELOG(" CertKey:    %s\n", _params.certKey);
 	WRITELOG(" PrivateKey: %s\n\n\n", _params.privateKey);
 
+	_stopFlg = false;
 
 	/* Run Tasks until CTRL+C entred */
 	MultiTaskProcess::run();
 
+	_stopFlg = true;
+
 	/* stop Tasks */
 	Event* ev = new Event();
 	ev->setStop();
@@ -274,10 +332,15 @@ void Gateway::run(void)
 	/* wait until all Task stop */
 	MultiTaskProcess::waitStop();
 
-	WRITELOG("\n%s MQTT-SN Gateway  stoped\n\n", currentDateTime());
+	WRITELOG("\n\n%s MQTT-SN Gateway  stopped.\n\n", currentDateTime());
 	_lightIndicator.allLightOff();
 }
 
+bool Gateway::IsStopping(void)
+{
+	return _stopFlg;
+}
+
 EventQue* Gateway::getPacketEventQue()
 {
 	return &_packetEventQue;

MQTTSNGateway/src/MQTTSNGateway.h

@@ -26,7 +26,7 @@ namespace MQTTSNGW
 /*=================================
  *    Starting prompt
  ==================================*/
-#define PAHO_COPYRIGHT0 " * MQTT-SN Transparent Gateway"
+#define PAHO_COPYRIGHT0 " * MQTT-SN Gateway"
 #define PAHO_COPYRIGHT1 " * Part of Project Paho in Eclipse"
 #define PAHO_COPYRIGHT2 " * (http://git.eclipse.org/c/paho/org.eclipse.paho.mqtt-sn.embedded-c.git/)"
 #define PAHO_COPYRIGHT3 " * Author : Tomoaki YAMAGUCHI"
@@ -160,10 +160,14 @@ public:
 	char* rootCApath {nullptr};
 	char* rootCAfile {nullptr};
 	char* certKey {nullptr};
-	char* privateKey {nullptr};
 	char* predefinedTopicFileName {nullptr};
+	char* privateKey {nullptr};
 	char* qosMinusClientListName {nullptr};
 	bool  clientAuthentication {false};
+	bool  predefinedTopic  {false};
+	bool  aggregatingGw   {false};
+	bool  qosMinus1   {false};
+	bool  forwarder   {false};
 };
 
 
@@ -190,8 +194,12 @@ public:
 	GatewayParams* getGWParams(void);
 	AdapterManager* getAdapterManager(void);
 	int getParam(const char* parameter, char* value);
+	char* getClientListFileName(void);
+	char* getPredefinedTopicFileName(void);
+
 	bool hasSecureConnection(void);
 	Topics* getTopics(void);
+	bool IsStopping(void);
 
 private:
 	GatewayParams  _params;
@@ -203,6 +211,7 @@ private:
 	SensorNetwork  _sensorNetwork;
 	AdapterManager* _adapterManager {nullptr};
 	Topics* _topics;
+	bool _stopFlg;
 };
 
 }

MQTTSNGateway/src/linux/Network.cpp

@@ -145,7 +145,11 @@ bool TCPStack::accept(TCPStack& new_socket)
 
 int TCPStack::send(const uint8_t* buf, int length)
 {
+#ifdef __APPLE__
+	return ::send(_sockfd, buf, length, SO_NOSIGPIPE);
+#else
 	return ::send(_sockfd, buf, length, MSG_NOSIGNAL);
+#endif
 }
 
 int TCPStack::recv(uint8_t* buf, int len)

MQTTSNGateway/src/linux/Threading.cpp

@@ -21,27 +21,40 @@
 #include <sys/ipc.h>
 #include <sys/shm.h>
 #include <sys/stat.h>
-#include <semaphore.h>
 #include <fcntl.h>
 #include <string.h>
 #include <pthread.h>
 #include <unistd.h>
+#include <errno.h>
 
 using namespace std;
 using namespace MQTTSNGW;
 
-#if defined(OSX)
-int sem_timedwait(sem_type sem, const struct timespec *timeout)
+#ifdef __APPLE__
+
+int sem_timedwait(sem_t *sem, const struct timespec *abs_timeout)
 {
-	int rc = -1;
-	int64_t tout = timeout->tv_sec * 1000L + tv_nsec * 1000000L
-	rc = (int)dispatch_semaphore_wait(sem, dispatch_time(DISPATCH_TIME_NOW, tout));
-	if (rc != 0)
+	while (true)
 	{
-		rc = ETIMEDOUT;
+		// try to lock the semaphore
+		int result = sem_trywait(sem);
+		if (result != -1 || errno != EAGAIN)
+			return result;
+
+		// spin lock
+		sched_yield();
+
+		// check if timeout reached
+		struct timespec ts;
+		clock_gettime(CLOCK_REALTIME, &ts);
+		if (ts.tv_sec > abs_timeout->tv_sec
+			|| (ts.tv_sec == abs_timeout->tv_sec && abs_timeout->tv_nsec >= ts.tv_nsec))
+		{
+			return ETIMEDOUT;
+		}
 	}
- 	return rc;
 }
+
 #endif
 
 /*=====================================
@@ -69,7 +82,7 @@ Mutex::Mutex(const char* fileName)
 		throw Exception( -1, "Mutex can't create a shared memory.");
 	}
 	_pmutex = (pthread_mutex_t*) shmat(_shmid, NULL, 0);
-	if (_pmutex < 0)
+	if (_pmutex == (void*) -1)
 	{
 		throw Exception( -1, "Mutex can't attach shared memory.");
 	}
@@ -153,21 +166,61 @@ void Mutex::unlock(void)
  Class Semaphore
  =====================================*/
 
-Semaphore::Semaphore()
-{
-	sem_init(&_sem, 0, 0);
-	_name = 0;
-	_psem = 0;
-}
-
 Semaphore::Semaphore(unsigned int val)
 {
+#ifdef __APPLE__
+	_sem = dispatch_semaphore_create(val);
+#else
 	sem_init(&_sem, 0, val);
-	_name = 0;
-	_psem = 0;
+#endif
 }
 
-Semaphore::Semaphore(const char* name, unsigned int val)
+Semaphore::~Semaphore()
+{
+#ifdef __APPLE__
+	dispatch_release(_sem);
+#else
+	sem_destroy(&_sem);
+#endif
+}
+
+void Semaphore::post(void)
+{
+#ifdef __APPLE__
+	dispatch_semaphore_signal(_sem);
+#else
+	sem_post(&_sem);
+#endif
+}
+
+void Semaphore::wait(void)
+{
+#ifdef __APPLE__
+	dispatch_semaphore_wait(_sem, DISPATCH_TIME_FOREVER);
+#else
+	sem_wait(&_sem);
+#endif
+}
+
+void Semaphore::timedwait(uint16_t millsec)
+{
+#ifdef __APPLE__
+	dispatch_semaphore_wait(_sem, dispatch_time(DISPATCH_TIME_NOW, int64_t(millsec) * 1000000));
+#else
+	struct timespec ts;
+	clock_gettime(CLOCK_REALTIME, &ts);
+	int nsec = ts.tv_nsec + (millsec % 1000) * 1000000;
+	ts.tv_nsec = nsec % 1000000000;
+	ts.tv_sec += millsec / 1000 + nsec / 1000000000;
+	sem_timedwait(&_sem, &ts);
+#endif
+}
+
+/*=====================================
+ Class NamedSemaphore
+ =====================================*/
+
+NamedSemaphore::NamedSemaphore(const char* name, unsigned int val)
 {
 	_psem = sem_open(name, O_CREAT, 0666, val);
 	if (_psem == SEM_FAILED)
@@ -181,67 +234,30 @@ Semaphore::Semaphore(const char* name, unsigned int val)
 	}
 }
 
-Semaphore::~Semaphore()
+NamedSemaphore::~NamedSemaphore()
 {
-	if (_name)
-	{
-		sem_close(_psem);
-		sem_unlink(_name);
-		free(_name);
-	}
-	else
-	{
-		sem_destroy(&_sem);
-	}
+	sem_close(_psem);
+	sem_unlink(_name);
+	free(_name);
 }
 
-void Semaphore::post(void)
+void NamedSemaphore::post(void)
 {
-	int val = 0;
-	if (_psem)
-	{
-		sem_getvalue(_psem, &val);
-		if (val <= 0)
-		{
-			sem_post(_psem);
-		}
-	}
-	else
-	{
-		sem_getvalue(&_sem, &val);
-		if (val <= 0)
-		{
-			sem_post(&_sem);
-		}
-	}
+	sem_post(_psem);
 }
 
-void Semaphore::wait(void)
+void NamedSemaphore::wait(void)
 {
-	if (_psem)
-	{
-		sem_wait(_psem);
-	}
-	else
-	{
-		sem_wait(&_sem);
-	}
+	sem_wait(_psem);
 }
 
-void Semaphore::timedwait(uint16_t millsec)
+void NamedSemaphore::timedwait(uint16_t millsec)
 {
 	struct timespec ts;
 	clock_gettime(CLOCK_REALTIME, &ts);
 	ts.tv_sec += millsec / 1000;
 	ts.tv_nsec = (millsec % 1000) * 1000000;
-	if (_psem)
-	{
-		sem_timedwait(_psem, &ts);
-	}
-	else
-	{
-		sem_timedwait(&_sem, &ts);
-	}
+	sem_timedwait(_psem, &ts);
 }
 
 /*=========================================
@@ -274,7 +290,7 @@ RingBuffer::RingBuffer(const char* keyDirectory)
 	if ((_shmid = shmget(key, PROCESS_LOG_BUFFER_SIZE,
 	IPC_CREAT | IPC_EXCL | 0666)) >= 0)
 	{
-		if ((_shmaddr = (uint16_t*) shmat(_shmid, NULL, 0)) > 0)
+		if ((_shmaddr = (uint16_t*) shmat(_shmid, NULL, 0)) != (void*) -1)
 		{
 			_length = (uint16_t*) _shmaddr;
 			_start = (uint16_t*) _length + sizeof(uint16_t*);
@@ -290,9 +306,9 @@ RingBuffer::RingBuffer(const char* keyDirectory)
 			throw Exception(-1, "RingBuffer can't attach shared memory.");
 		}
 	}
-	else if ((_shmid = shmget(key, PROCESS_LOG_BUFFER_SIZE, IPC_CREAT | 0666)) >= 0)
+	else if ((_shmid = shmget(key, PROCESS_LOG_BUFFER_SIZE, IPC_CREAT | 0666)) != -1)
 	{
-		if ((_shmaddr = (uint16_t*) shmat(_shmid, NULL, 0)) > 0)
+		if ((_shmaddr = (uint16_t*) shmat(_shmid, NULL, 0)) != (void*) -1)
 		{
 			_length = (uint16_t*) _shmaddr;
 			_start = (uint16_t*) _length + sizeof(uint16_t*);
@@ -330,7 +346,7 @@ RingBuffer::~RingBuffer()
 		}
 	}
 
-	if (_pmx > 0)
+	if (_pmx != NULL)
 	{
 		delete _pmx;
 	}
@@ -525,7 +541,7 @@ int Thread::start(void)
 
 void Thread::stopProcess(void)
 {
-	theMultiTaskProcess->threadStoped();
+	theMultiTaskProcess->threadStopped();
 }
 
 void Thread::stop(void)

MQTTSNGateway/src/linux/Threading.h

@@ -19,6 +19,9 @@
 
 #include <pthread.h>
 #include <semaphore.h>
+#ifdef __APPLE__
+#include <dispatch/dispatch.h>
+#endif
 #include "MQTTSNGWDefines.h"
 
 namespace MQTTSNGW
@@ -52,17 +55,34 @@ private:
 class Semaphore
 {
 public:
-	Semaphore();
-	Semaphore(unsigned int val);
-	Semaphore(const char* name, unsigned int val);
+	Semaphore(unsigned int val = 0);
 	~Semaphore();
 	void post(void);
 	void wait(void);
 	void timedwait(uint16_t millsec);
 
+private:
+#ifdef __APPLE__
+	dispatch_semaphore_t _sem;
+#else
+	sem_t _sem;
+#endif
+};
+
+/*=====================================
+         Class NamedSemaphore
+  ====================================*/
+class NamedSemaphore
+{
+public:
+	NamedSemaphore(const char* name, unsigned int val);
+	~NamedSemaphore();
+	void post(void);
+	void wait(void);
+	void timedwait(uint16_t millsec);
+
 private:
 	sem_t* _psem;
-	sem_t  _sem;
 	char*  _name;
 };
 

MQTTSNGateway/src/linux/loralink/SensorNetwork.cpp

@@ -0,0 +1,559 @@
+/**************************************************************************************
+ * 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.
+ *
+ * Contributors:
+ *    Tomoaki Yamaguchi - initial API and implementation 
+ **************************************************************************************/
+
+#include <stdio.h>
+#include <unistd.h>
+#include <termios.h>
+#include <fcntl.h>
+#include <string.h>
+#include <stdlib.h>
+#include <sys/time.h>
+#include <sys/types.h>
+#include <sys/select.h>
+#include <errno.h>
+#include "SensorNetwork.h"
+#include "MQTTSNGWProcess.h"
+
+using namespace std;
+using namespace MQTTSNGW;
+
+#define LORA_PHY_MAXPAYLOAD      256
+#define LORALINK_MAX_API_LEN     ( LORA_PHY_MAXPAYLOAD + 5 )  // PayloadType[1] + Rssi[2] + Snr[2]
+
+#define LORALINK_ACK          0x10
+#define LORALINK_NO_FREE_CH   0x20
+#define LORALINK_TX_TIMEOUT   0x40
+
+#define LORALINK_TIMEOUT_ACK  10000      // 10 secs
+
+/*===========================================
+ Class  SensorNetAddreess
+ ============================================*/
+SensorNetAddress::SensorNetAddress()
+{
+	_devAddr = 0;
+}
+
+SensorNetAddress::~SensorNetAddress()
+{
+
+}
+
+void SensorNetAddress::setAddress( uint8_t devAddr)
+{
+	_devAddr = devAddr;
+}
+
+
+int SensorNetAddress::setAddress(string* address)
+{
+	_devAddr = atoi(address->c_str());
+
+	if ( _devAddr == 0 )
+	{
+		return -1;
+	}
+	return 0;
+}
+
+void SensorNetAddress::setBroadcastAddress(void)
+{
+	_devAddr = BROADCAST_DEVADDR;
+}
+
+bool SensorNetAddress::isMatch(SensorNetAddress* addr)
+{
+	return _devAddr == addr->_devAddr;
+}
+
+SensorNetAddress& SensorNetAddress::operator =(SensorNetAddress& addr)
+{
+	_devAddr =  addr._devAddr;
+	return *this;
+}
+
+char* SensorNetAddress::sprint(char* buf)
+{
+		sprintf( buf, "%d", _devAddr);
+	return buf;
+}
+
+/*===========================================
+ Class  SensorNetwork
+ ============================================*/
+SensorNetwork::SensorNetwork()
+{
+
+}
+
+SensorNetwork::~SensorNetwork()
+{
+
+}
+
+int SensorNetwork::unicast(const uint8_t* payload, uint16_t payloadLength, SensorNetAddress* sendToAddr)
+{
+	return LoRaLink::unicast(payload, payloadLength, sendToAddr);
+}
+
+int SensorNetwork::broadcast(const uint8_t* payload, uint16_t payloadLength)
+{
+	return LoRaLink::broadcast(payload, payloadLength);
+}
+
+int SensorNetwork::read(uint8_t* buf, uint16_t bufLen)
+{
+	return LoRaLink::recv(buf, bufLen, &_clientAddr);
+}
+
+int SensorNetwork::initialize(void)
+{
+	char param[MQTTSNGW_PARAM_MAX];
+	uint32_t baudrate = 115200;
+
+	if (theProcess->getParam("BaudrateLoRaLink", param) == 0)
+	{
+		baudrate = (uint32_t)atoi(param);
+	}
+	_description += "LoRaLink, Baudrate ";
+	sprintf(param ,"%d", baudrate);
+	_description += param;
+
+	theProcess->getParam("DeviceRxLoRaLink", param);
+	_description += ", SerialRx ";
+	_description += param;
+	if ( LoRaLink::open(LORALINK_MODEM_RX, param, baudrate) < 0 )
+	{
+		return -1;
+	}
+
+	theProcess->getParam("DeviceTxLoRaLink", param);
+	_description += ", SerialTx ";
+	_description += param;
+	return LoRaLink::open(LORALINK_MODEM_TX, param, baudrate);
+}
+
+const char* SensorNetwork::getDescription(void)
+{
+	return _description.c_str();
+}
+
+SensorNetAddress* SensorNetwork::getSenderAddress(void)
+{
+	return &_clientAddr;
+}
+
+/*===========================================
+              Class  LoRaLink
+ ============================================*/
+LoRaLink::LoRaLink(){
+    _serialPortRx = new SerialPort();
+    _serialPortTx = new SerialPort();
+    _respCd = 0;
+}
+
+LoRaLink::~LoRaLink(){
+	if ( _serialPortRx )
+	{
+		delete _serialPortRx;
+	}
+	if ( _serialPortTx )
+	{
+		delete _serialPortTx;
+	}
+}
+
+int LoRaLink::open(LoRaLinkModemType_t type, char* device, int baudrate)
+{
+	int rate = B9600;
+
+	switch (baudrate)
+	{
+	case 9600:
+		rate = B9600;
+		break;
+	case 19200:
+		rate = B19200;
+		break;
+	case 38400:
+		rate = B38400;
+		break;
+	case 57600:
+		rate = B57600;
+		break;
+	case 115200:
+		rate = B115200;
+		break;
+	default:
+		return -1;
+	}
+
+	int rc = 0;
+	if ( type == LORALINK_MODEM_RX )
+	{
+		if (  (rc = _serialPortRx->open(device, rate, false, 1, O_RDWR | O_NOCTTY)) < 0 )
+		return rc;
+	}
+	else
+	{
+		rc =  _serialPortTx->open(device, rate, false, 1, O_RDWR | O_NOCTTY);
+	}
+	return rc;
+}
+
+int LoRaLink::broadcast(const uint8_t* payload, uint16_t payloadLen){
+	SensorNetAddress addr;
+	addr.setBroadcastAddress();
+	return send(MQTT_SN, payload, (uint8_t) payloadLen, &addr);
+}
+
+int LoRaLink:: unicast(const uint8_t* payload, uint16_t payloadLen, SensorNetAddress* addr){
+	return send(MQTT_SN, payload, (uint8_t) payloadLen, addr);
+}
+
+int LoRaLink::recv(uint8_t* buf, uint16_t bufLen, SensorNetAddress* clientAddr)
+{
+	while ( true )
+	{
+		if ( ( readApiFrame( &_loRaLinkApi, &_loRaLinkPara) == true ) && (_loRaLinkPara.Available == true) && ( _loRaLinkPara.Error == false ) )
+		{
+			clientAddr->_devAddr = _loRaLinkApi.SourceAddr;
+
+			bufLen = _loRaLinkApi.PayloadLen;
+
+			memcpy( buf, _loRaLinkApi.Payload, bufLen );
+
+			switch ( (int) _loRaLinkApi.PayloadType )
+			{
+			case API_RSP_ACK:
+				_respCd = LORALINK_ACK;
+				break;
+
+			case API_RSP_NFC:
+				_respCd = LORALINK_NO_FREE_CH;
+				break;
+
+			case API_RSP_TOT:
+				_respCd = LORALINK_TX_TIMEOUT;
+				break;
+
+
+			case MQTT_SN:
+				memcpy( buf, _loRaLinkApi.Payload, bufLen );
+				return bufLen;
+
+			default:
+				return 0;
+			}
+			_sem.post();
+			return bufLen;
+
+		}
+		else
+		{
+		    return 0;
+		}
+	}
+}
+
+
+
+bool LoRaLink::readApiFrame(LoRaLinkFrame_t* api, LoRaLinkReadParameters_t* para)
+{
+	uint8_t byte = 0;
+	uint16_t val = 0;
+
+	while ( recv(&byte) == 0 )
+	{
+		if ( byte == START_BYTE )
+		{
+			para->apipos= 1;
+			para->Error = true;
+			para->Available = false;
+			continue;
+		}
+
+		if ( para->apipos > 0 && byte == ESCAPE )
+		{
+			if( recv(&byte ) == 0 )
+			{
+				byte ^= PAD;  // decode
+			}
+			else
+			{
+				para->Escape = true;
+				continue;
+			}
+		}
+
+		if( para->Escape == true )
+		{
+			byte ^= PAD;
+			para->Escape = false;
+		}
+
+		switch ( para->apipos )
+		{
+		case 0:
+			break;
+
+		case 1:
+			val = (uint16_t)byte;
+			api->PayloadLen = val << 8;
+			break;
+
+		case 2:
+			api->PayloadLen += byte;
+			break;
+
+		case 3:
+			api->SourceAddr = byte;
+			para->checksum = byte;
+			break;
+
+		case 4:
+			val = (uint16_t)byte;
+			api->Rssi = val << 8;
+			para->checksum += byte;
+			break;
+
+		case 5:
+			api->Rssi += byte;
+			para->checksum += byte;
+			break;
+
+		case 6:
+			val = (uint16_t)byte;
+			api->Snr = val << 8;
+			para->checksum += byte;
+			break;
+
+		case 7:
+			api->Snr += byte;
+			para->checksum += byte;
+			break;
+
+		case 8:
+			api->PayloadType = byte;
+			para->checksum += byte;
+			break;
+
+		default:
+			if ( para->apipos >= api->PayloadLen + 2 )    //  FRM_DEL + CRC = 2
+			{
+				para->Error = ( (0xff - para->checksum) != byte );
+				para->Available = true;
+				api->PayloadLen -= 7;   // 7 = SrcAddr[1] + Rssi[2] + Snr[2] + PlType[1] + Crc[1]
+				para->apipos = 0;
+				para->checksum = 0;
+				return true;
+			}
+			else
+			{
+				para->checksum += byte;
+				api->Payload[para->apipos - 9] = byte;
+			}
+			break;
+		}
+
+		para->apipos++;
+	}
+	return false;
+}
+
+int LoRaLink::send(LoRaLinkPayloadType_t type, const uint8_t* payload, uint16_t pLen, SensorNetAddress* addr)
+{
+	D_NWSTACK("\r\n===> Send:    ");
+	uint8_t buf[2] = { 0 };
+	uint8_t chks = 0;
+	uint16_t len = pLen + 3;   // 3 = DestAddr[1] + PayloadType[1] + Crc[1]
+	_respCd = 0;
+
+    _serialPortTx->send(START_BYTE);
+
+    buf[0] = (len >> 8) & 0xff;
+    buf[1] = len & 0xff;
+    send(buf[0]);
+    send(buf[1]);
+
+    send( addr->_devAddr );
+    chks = addr->_devAddr;
+
+
+
+    send(type);
+    chks += type;
+
+    D_NWSTACK("\r\n     Payload: ");
+
+    for ( uint8_t i = 0; i < pLen; i++ ){
+        send(payload[i]);     // Payload
+        chks += payload[i];
+    }
+
+    chks = 0xff - chks;
+    D_NWSTACK("   checksum  ");
+    send(chks);
+    D_NWSTACK("\r\n");
+
+    /* wait ACK */
+    _sem.timedwait(LORALINK_TIMEOUT_ACK);
+
+    if ( _respCd == LORALINK_NO_FREE_CH )
+    {
+    	 D_NWSTACK(" Channel isn't free\r\n");
+    	return -1;
+    }
+    else if ( _respCd != LORALINK_ACK )
+	{
+		 D_NWSTACK(" Not Acknowleged\r\n");
+		return -1;
+	}
+    return (int)pLen;
+}
+
+void LoRaLink::send(uint8_t c)
+{
+  if( (c == START_BYTE || c == ESCAPE || c == XON || c == XOFF)){
+	  _serialPortTx->send(ESCAPE);
+	  _serialPortTx->send(c ^ PAD);
+  }else{
+	  _serialPortTx->send(c);
+  }
+}
+
+int LoRaLink::recv(uint8_t* buf)
+{
+    struct timeval timeout;
+    int maxfd;
+    int fd;
+    fd_set rfds;
+
+    timeout.tv_sec = 1;
+    timeout.tv_usec = 0;    // 500ms
+    FD_ZERO(&rfds);
+    FD_SET(_serialPortRx->_fd, &rfds);
+    FD_SET(_serialPortTx->_fd, &rfds);
+    if ( _serialPortRx->_fd > _serialPortTx->_fd )
+    {
+    	maxfd = _serialPortRx->_fd;
+    }
+    else
+    {
+    	maxfd = _serialPortTx->_fd;
+    }
+
+    if ( select(maxfd + 1, &rfds, 0, 0, &timeout) > 0 )
+    {
+    	if ( FD_ISSET(_serialPortRx->_fd, &rfds) )
+		{
+    		fd = _serialPortRx->_fd;
+		}
+    	else
+		{
+    		fd = _serialPortTx->_fd;
+		}
+
+		if ( read(fd, buf, 1) == 1 )
+		{
+			/*
+			if ( *buf == ESCAPE )
+			{
+				D_NWSTACK( " %02x",buf[0] );
+				if ( read(fd, buf, 1) == 1 )
+				{
+					*buf = PAD ^ *buf;
+				}
+				else
+				{
+					return -1;
+				}
+
+			}
+			*/
+			D_NWSTACK( " %02x",buf[0] );
+			return 0;
+		}
+	}
+    return -1;
+}
+
+/*=========================================
+ Class SerialPort
+ =========================================*/
+SerialPort::SerialPort()
+{
+	_tio.c_iflag = IGNBRK | IGNPAR;
+	_tio.c_cflag = CS8 | CLOCAL | CREAD;
+	_tio.c_cc[VINTR] = 0;
+	_tio.c_cc[VTIME] = 10;   // 1 sec.
+	_tio.c_cc[VMIN] = 1;
+	_fd = 0;
+}
+
+SerialPort::~SerialPort()
+{
+	if (_fd)
+	{
+		::close(_fd);
+	}
+}
+
+int SerialPort::open(char* devName, unsigned int baudrate, bool parity,
+		unsigned int stopbit, unsigned int flg)
+{
+	_fd = ::open(devName, flg);
+	if (_fd < 0)
+	{
+		return _fd;
+	}
+
+	if (parity)
+	{
+		_tio.c_cflag = _tio.c_cflag | PARENB;
+	}
+	if (stopbit == 2)
+	{
+		_tio.c_cflag = _tio.c_cflag | CSTOPB;
+	}
+
+	if (cfsetspeed(&_tio, baudrate) < 0)
+	{
+		return errno;
+	}
+	return tcsetattr(_fd, TCSANOW, &_tio);
+}
+
+bool SerialPort::send(unsigned char b)
+{
+	if (write(_fd, &b, 1) <= 0)
+	{
+		return false;
+	}
+	else
+	{
+		D_NWSTACK( " %02x", b);
+		return true;
+	}
+}
+
+
+
+void SerialPort::flush(void)
+{
+	tcsetattr(_fd, TCSAFLUSH, &_tio);
+}
+

MQTTSNGateway/src/linux/loralink/SensorNetwork.h

@@ -0,0 +1,187 @@
+/**************************************************************************************
+ * 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.
+ *
+ * Contributors:
+ *    Tomoaki Yamaguchi - initial API and implementation 
+ **************************************************************************************/
+#ifndef SENSORNETWORKX_H_
+#define SENSORNETWORKX_H_
+
+#include "MQTTSNGWDefines.h"
+#include "MQTTSNGWProcess.h"
+#include <string>
+#include <termios.h>
+
+using namespace std;
+
+namespace MQTTSNGW
+{
+//#define DEBUG_NWSTACK
+
+#ifdef  DEBUG_NWSTACK
+  #define D_NWSTACK(...) printf(__VA_ARGS__); fflush(stdout)
+#else
+  #define D_NWSTACK(...)
+#endif
+
+
+#define XMIT_STATUS_TIME_OVER    5000
+
+#define START_BYTE               0x7e
+#define ESCAPE                   0x7d
+#define XON                      0x11
+#define XOFF                     0x13
+#define PAD                      0x20
+
+#define BROADCAST_DEVADDR        0xFF
+
+#define LORA_PHY_MAXPAYLOAD      256
+
+/*!
+ * LoRaLink Modem Type
+ */
+typedef enum
+{
+	LORALINK_MODEM_TX,
+	LORALINK_MODEM_RX,
+}LoRaLinkModemType_t;
+
+/*!
+ * LoRaLink Serialized API
+ */
+typedef struct
+{
+	uint16_t  PanId;
+	uint8_t   DestinationAddr;
+	uint8_t   SourceAddr;
+	uint16_t  Rssi;
+	uint16_t  Snr;
+	uint8_t   PayloadType;
+	uint8_t   Payload[LORA_PHY_MAXPAYLOAD];
+	uint16_t  PayloadLen;
+}LoRaLinkFrame_t;
+
+typedef struct
+{
+	bool Available;
+	bool Error;
+	bool Escape;
+	uint16_t apipos;
+	uint8_t checksum;
+} LoRaLinkReadParameters_t;
+
+typedef enum
+{
+	MQTT_SN        = 0x40,
+	API_RSP_ACK    = 0x80,
+	API_RSP_NFC,
+	API_RSP_TOT,
+	API_REQ_UTC,
+	API_RSP_UTC,
+	API_CHG_TASK_PARAM,
+	API_REQ_RESET,
+
+}LoRaLinkPayloadType_t;
+
+/*===========================================
+  Class  SerialPort
+ ============================================*/
+class SerialPort
+{
+	friend class LoRaLink;
+public:
+	SerialPort();
+	~SerialPort();
+	int open(char* devName, unsigned int baudrate,  bool parity, unsigned int stopbit, unsigned int flg);
+	bool send(unsigned char b);
+	void flush();
+
+private:
+	int _fd;  // file descriptor
+	struct termios _tio;
+};
+
+/*===========================================
+ Class  SensorNetAddreess
+ ============================================*/
+class SensorNetAddress
+{
+	friend class LoRaLink;
+public:
+	SensorNetAddress();
+	~SensorNetAddress();
+	void setAddress( uint8_t devAddr);
+	int  setAddress(string* data);
+	void setBroadcastAddress(void);
+	bool isMatch(SensorNetAddress* addr);
+	SensorNetAddress& operator =(SensorNetAddress& addr);
+	char* sprint(char*);
+private:
+	uint8_t _devAddr;
+//	uint8_t _destAddr;
+};
+
+/*========================================
+ Class LoRaLink
+ =======================================*/
+class LoRaLink
+{
+public:
+	LoRaLink();
+	~LoRaLink();
+
+	int open(LoRaLinkModemType_t type, char* device, int boudrate );
+	void close(void);
+	int unicast(const uint8_t* buf, uint16_t length, SensorNetAddress* sendToAddr);
+	int broadcast(const uint8_t* buf, uint16_t length);
+	int recv(uint8_t* buf, uint16_t len, SensorNetAddress* addr);
+	void setApiMode(uint8_t mode);
+
+private:
+	bool readApiFrame(LoRaLinkFrame_t* api, LoRaLinkReadParameters_t* para);
+	int recv(uint8_t* buf);
+	int send(LoRaLinkPayloadType_t type, const uint8_t* payload, uint16_t pLen, SensorNetAddress* addr);
+	void send(uint8_t b);
+
+	Semaphore _sem;
+	Mutex _meutex;
+	uint8_t _respCd;
+	SerialPort* _serialPortRx;
+	SerialPort* _serialPortTx;
+	LoRaLinkFrame_t _loRaLinkApi;
+	LoRaLinkReadParameters_t _loRaLinkPara;
+};
+
+/*===========================================
+ Class  SensorNetwork
+ ============================================*/
+class SensorNetwork: public LoRaLink
+{
+public:
+	SensorNetwork();
+	~SensorNetwork();
+
+	int unicast(const uint8_t* payload, uint16_t payloadLength, SensorNetAddress* sendto);
+	int broadcast(const uint8_t* payload, uint16_t payloadLength);
+	int read(uint8_t* buf, uint16_t bufLen);
+	int initialize(void);
+	const char* getDescription(void);
+	SensorNetAddress* getSenderAddress(void);
+
+private:
+	SensorNetAddress _clientAddr;   // Sender's address. not gateway's one.
+	string _description;
+};
+
+}
+
+#endif /* SENSORNETWORKX_H_ */

MQTTSNGateway/src/linux/udp/SensorNetwork.cpp

@@ -185,7 +185,7 @@ int SensorNetwork::initialize(void)
 	uint16_t multicastPortNo = 0;
 	uint16_t unicastPortNo = 0;
 	string ip;
-
+	unsigned int ttl = 1;
 	/*
 	 * theProcess->getParam( ) copies
 	 * a text specified by "Key" into param[] from the Gateway.conf
@@ -216,9 +216,15 @@ int SensorNetwork::initialize(void)
 		_description += " Gateway Port ";
 		_description += param;
 	}
+	if (theProcess->getParam("MulticastTTL", param) == 0)
+	{
+		ttl = atoi(param);
+		_description += " TTL: ";
+		_description += param;
+	}
 
 	/*  Prepare UDP sockets */
-	return UDPPort::open(ip.c_str(), multicastPortNo, unicastPortNo);
+	return UDPPort::open(ip.c_str(), multicastPortNo, unicastPortNo, ttl);
 }
 
 const char* SensorNetwork::getDescription(void)
@@ -261,7 +267,7 @@ void UDPPort::close(void)
 	}
 }
 
-int UDPPort::open(const char* ipAddress, uint16_t multiPortNo, uint16_t uniPortNo)
+int UDPPort::open(const char* ipAddress, uint16_t multiPortNo, uint16_t uniPortNo, unsigned int ttl)
 {
 	char loopch = 0;
 	const int reuse = 1;
@@ -275,6 +281,7 @@ int UDPPort::open(const char* ipAddress, uint16_t multiPortNo, uint16_t uniPortN
 	uint32_t ip = inet_addr(ipAddress);
 	_grpAddr.setAddress(ip, htons(multiPortNo));
 	_clientAddr.setAddress(ip, htons(uniPortNo));
+	_ttl = ttl;
 
 	/*------ Create unicast socket --------*/
 	_sockfdUnicast = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
@@ -342,6 +349,13 @@ int UDPPort::open(const char* ipAddress, uint16_t multiPortNo, uint16_t uniPortN
 		return -1;
 	}
 
+	if (setsockopt(_sockfdMulticast, IPPROTO_IP, IP_MULTICAST_TTL, &ttl,sizeof(ttl)) < 0)
+	{
+		D_NWSTACK("error Multicast IP_ADD_MEMBERSHIP in UDPPort::open\n");
+		close();
+		return -1;
+	}
+
 	if (setsockopt(_sockfdUnicast, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq)) < 0)
 	{
 		D_NWSTACK("error Unicast IP_ADD_MEMBERSHIP in UDPPort::open\n");
@@ -378,8 +392,8 @@ int UDPPort::recv(uint8_t* buf, uint16_t len, SensorNetAddress* addr)
 	fd_set recvfds;
 	int maxSock = 0;
 
-	timeout.tv_sec = 0;
-	timeout.tv_usec = 1000000;    // 1 sec
+	timeout.tv_sec = 1;
+	timeout.tv_usec = 0;    // 1 sec
 	FD_ZERO(&recvfds);
 	FD_SET(_sockfdUnicast, &recvfds);
 	FD_SET(_sockfdMulticast, &recvfds);

MQTTSNGateway/src/linux/udp/SensorNetwork.h

@@ -60,7 +60,7 @@ public:
 	UDPPort();
 	virtual ~UDPPort();
 
-	int open(const char* ipAddress, uint16_t multiPortNo,	uint16_t uniPortNo);
+	int open(const char* ipAddress, uint16_t multiPortNo,	uint16_t uniPortNo, unsigned int hops);
 	void close(void);
 	int unicast(const uint8_t* buf, uint32_t length, SensorNetAddress* sendToAddr);
 	int broadcast(const uint8_t* buf, uint32_t length);
@@ -76,7 +76,7 @@ private:
 	SensorNetAddress _grpAddr;
 	SensorNetAddress _clientAddr;
 	bool _disconReq;
-
+	unsigned int _ttl;
 };
 
 /*===========================================

MQTTSNGateway/src/linux/udp6/SensorNetwork.cpp

@@ -122,11 +122,8 @@ char* SensorNetAddress::getAddress(void)
 
 bool SensorNetAddress::isMatch(SensorNetAddress* addr)
 {
-	return ((this->_portNo == addr->_portNo) && \
-	(this->_IpAddr.sin6_addr.s6_addr32[0] == addr->_IpAddr.sin6_addr.s6_addr32[0]) && \
-	(this->_IpAddr.sin6_addr.s6_addr32[1] == addr->_IpAddr.sin6_addr.s6_addr32[1]) && \
-	(this->_IpAddr.sin6_addr.s6_addr32[2] == addr->_IpAddr.sin6_addr.s6_addr32[2]) && \
-	(this->_IpAddr.sin6_addr.s6_addr32[3] == addr->_IpAddr.sin6_addr.s6_addr32[3]));
+	return (this->_portNo == addr->_portNo) && \
+	(memcmp(this->_IpAddr.sin6_addr.s6_addr, addr->_IpAddr.sin6_addr.s6_addr, sizeof(this->_IpAddr.sin6_addr.s6_addr)) == 0);
 }
 
 SensorNetAddress& SensorNetAddress::operator =(SensorNetAddress& addr)
@@ -179,6 +176,7 @@ int SensorNetwork::initialize(void)
 	string ip;
 	string broadcast;
 	string interface;
+	unsigned int hops = 1;
 
 	if (theProcess->getParam("GatewayUDP6Bind", param) == 0)
 	{
@@ -204,8 +202,14 @@ int SensorNetwork::initialize(void)
 		_description += " Interface: ";
 		_description += param;
 	}
+	if (theProcess->getParam("GatewayUDP6Hops", param) == 0)
+	{
+		hops = atoi(param);
+		_description += " Hops: ";
+		_description += param;
+	}
 
-	return UDPPort6::open(ip.c_str(), unicastPortNo, broadcast.c_str(), interface.c_str());
+	return UDPPort6::open(ip.c_str(), unicastPortNo, broadcast.c_str(), interface.c_str(), hops);
 }
 
 const char* SensorNetwork::getDescription(void)
@@ -248,7 +252,7 @@ void UDPPort6::close(void)
 	}
 }
 
-int UDPPort6::open(const char* ipAddress, uint16_t uniPortNo, const char* broadcastAddr, const char* interfaceName)
+int UDPPort6::open(const char* ipAddress, uint16_t uniPortNo, const char* broadcastAddr, const char* interfaceName, unsigned int hops)
 {
 	struct addrinfo hints, *res;
 	int errnu;
@@ -295,8 +299,15 @@ int UDPPort6::open(const char* ipAddress, uint16_t uniPortNo, const char* broadc
 		WRITELOG("UDP6::open - limit IPv6: %s",strerror(errnu));
 		return errnu;
 	}
+	errnu = setsockopt(_sockfdMulticast, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, &hops,sizeof(hops));
+	if(errnu <0)
+	{
+	    WRITELOG("UDP6::open - limit HOPS: %s",strerror(errnu));
+	    return errnu;
+	}
 
 	_uniPortNo = uniPortNo;
+	_hops = hops;
 	freeaddrinfo(res);
 
 	//init the structs for getaddrinfo
@@ -320,8 +331,13 @@ int UDPPort6::open(const char* ipAddress, uint16_t uniPortNo, const char* broadc
 	//if given, set a given device name to bind to
 	if(strlen(interfaceName) > 0)
 	{
+#ifdef __APPLE__
+		int idx = if_nametoindex(interfaceName);
+		setsockopt(_sockfdUnicast, IPPROTO_IP, IP_BOUND_IF, &idx, sizeof(idx));
+#else	
 		//socket option: bind to a given interface name
 		setsockopt(_sockfdUnicast, SOL_SOCKET, SO_BINDTODEVICE, interfaceName, strlen(interfaceName));
+#endif
 	}
 
 	//socket option: reuse address
@@ -371,7 +387,7 @@ int UDPPort6::unicast(const uint8_t* buf, uint32_t length, SensorNetAddress* add
 		strcpy(destStr, addr->getAddress());
 		strcat(destStr,"%");
 		strcat(destStr,_interfaceName);
-		if(IN6_IS_ADDR_LINKLOCAL(addr->getAddress()))
+		if(IN6_IS_ADDR_LINKLOCAL(&addr->getIpAddress()->sin6_addr))
 		{
 			getaddrinfo(destStr, portStr.c_str(), &hints, &res);
 		}
@@ -412,8 +428,8 @@ int UDPPort6::broadcast(const uint8_t* buf, uint32_t length)
 		strcpy(destStr, _grpAddr.getAddress());
 		strcat(destStr,"%");
 		strcat(destStr,_interfaceName);
-		if(IN6_IS_ADDR_MC_NODELOCAL(_grpAddr.getAddress()) ||
-		   IN6_IS_ADDR_MC_LINKLOCAL(_grpAddr.getAddress()))
+		if(IN6_IS_ADDR_MC_NODELOCAL(&_grpAddr.getIpAddress()->sin6_addr) ||
+		   IN6_IS_ADDR_MC_LINKLOCAL(&_grpAddr.getIpAddress()->sin6_addr))
 		{
 			err = getaddrinfo(destStr, std::to_string(_uniPortNo).c_str(), &hint, &info );
 		}
@@ -440,14 +456,13 @@ int UDPPort6::broadcast(const uint8_t* buf, uint32_t length)
 	return 0;
 }
 
-//TODO: test if this is working properly (GW works, but this function is not completely tested)
 int UDPPort6::recv(uint8_t* buf, uint16_t len, SensorNetAddress* addr)
 {
 	struct timeval timeout;
 	fd_set recvfds;
 
-	timeout.tv_sec = 0;
-	timeout.tv_usec = 1000000;    // 1 sec
+	timeout.tv_sec = 1;
+	timeout.tv_usec = 0;    // 1 sec
 	FD_ZERO(&recvfds);
 	FD_SET(_sockfdUnicast, &recvfds);
 
@@ -462,7 +477,6 @@ int UDPPort6::recv(uint8_t* buf, uint16_t len, SensorNetAddress* addr)
 	return rc;
 }
 
-//TODO: test if this is working properly (GW works, but this function is not completely tested)
 int UDPPort6::recvfrom(int sockfd, uint8_t* buf, uint16_t len, uint8_t flags, SensorNetAddress* addr)
 {
 	sockaddr_in6 sender;

MQTTSNGateway/src/linux/udp6/SensorNetwork.h

@@ -66,7 +66,7 @@ public:
 	UDPPort6();
 	virtual ~UDPPort6();
 
-	int open(const char* ipAddress, uint16_t uniPortNo, const char* broadcastAddr, const char* interfaceName);
+	int open(const char* ipAddress, uint16_t uniPortNo, const char* broadcastAddr, const char* interfaceName, unsigned int hops);
 	void close(void);
 	int unicast(const uint8_t* buf, uint32_t length, SensorNetAddress* sendToAddr);
 	int broadcast(const uint8_t* buf, uint32_t length);
@@ -84,7 +84,7 @@ private:
 	SensorNetAddress _clientAddr;
 	uint16_t _uniPortNo;
 	bool _disconReq;
-
+	unsigned int _hops;
 };
 
 /*===========================================

MQTTSNGateway/src/mainGateway.cpp

@@ -34,12 +34,17 @@ BrokerSendTask    task5(&gateway);
 
 int main(int argc, char** argv)
 {
-	try {
-		gateway.initialize(argc, argv);
-	    gateway.run();
-	} catch (const std::exception &ex) {
-		WRITELOG("\nEclipse Paho MQTT-SN Gateway exception: %s\n", ex.what());
-		WRITELOG("MQTT-SNGateway [-f Config file name]\n");
-	}
+    gateway.initialize(argc, argv);
+    gateway.run();
+    try
+    {
+      gateway.initialize(argc, argv);
+		  gateway.run();
+	  }
+    catch (const std::exception &ex)
+    {
+		  WRITELOG("\nEclipse Paho MQTT-SN Gateway exception: %s\n", ex.what());
+		  WRITELOG("MQTT-SNGateway [-f Config file name]\n");
+	  }
 	return 0;
 }

MQTTSNGateway/src/tests/TestTopics.cpp

@@ -214,7 +214,7 @@ void TestTopics::test(void)
 	for ( int i = 0; i < 10 ; i++ )
 	{
 		MQTTSN_topicid tp1;
-		char tp0[10];
+		char tp0[20];
 		sprintf(tp0, "Topic/%d/%d", i, i);
 		tp1.type = MQTTSN_TOPIC_TYPE_NORMAL;
 		tp1.data.long_.len = strlen(tp0);
@@ -237,7 +237,7 @@ void TestTopics::test(void)
 	for ( int i = 0; i < 10 ; i++ )
 	{
 		MQTTSN_topicid tp1;
-		char tp0[10];
+		char tp0[20];
 		sprintf(tp0, "Topic/%d", i);
 		tp1.type = MQTTSN_TOPIC_TYPE_NORMAL;
 		tp1.data.long_.len = strlen(tp0);
@@ -261,7 +261,7 @@ void TestTopics::test(void)
 	for ( int i = 0; i < 10 ; i++ )
 	{
 		MQTTSN_topicid tpid1;
-		char tp0[10];
+		char tp0[20];
 		sprintf(tp0, "TOPIC/%d/%d", i, i);
 		tpid1.type = MQTTSN_TOPIC_TYPE_NORMAL;
 		tpid1.data.long_.len = strlen(tp0);

MQTTSNPacket/src/MQTTSNPacket.c

@@ -53,7 +53,7 @@ const char* MQTTSNPacket_name(int code)
  */
 int MQTTSNPacket_len(int length)
 {
-	return (length > 255) ? length + 3 : length + 1;
+	return (length >= 255) ? length + 3 : length + 1;
 }
 
 /**
@@ -67,7 +67,7 @@ int MQTTSNPacket_encode(unsigned char* buf, int length)
 	int rc = 0;
 
 	FUNC_ENTRY;
-	if (length > 255)
+	if (length >= 255)
 	{
 		writeChar(&buf, 0x01);
 		writeInt(&buf, length);
@@ -83,7 +83,8 @@ int MQTTSNPacket_encode(unsigned char* buf, int length)
 
 /**
  * Obtains the MQTT-SN packet length from received data
- * @param getcharfn pointer to function to read the next character from the data source
+ * @param buf the buffer that contains the MQTT-SN packet
+ * @param buflen the length in bytes of the supplied buffer
  * @param value the decoded length returned
  * @return the number of bytes read from the socket
  */

MQTTSNPacket/src/MQTTSNSearchClient.c

@@ -116,7 +116,7 @@ int MQTTSNDeserialize_gwinfo(unsigned char* gatewayid, unsigned short* gatewayad
 	*gatewayid = readChar(&curdata);
 
 	*gatewayaddress_len = enddata - curdata;
-	*gatewayaddress = (gatewayaddress_len > 0) ? curdata : NULL;
+	*gatewayaddress = (*gatewayaddress_len > 0) ? curdata : NULL;
 
 	rc = 1;
 exit: