Merge pull request #143 from eclipse/develop

Add functions of Aggregating Gateway, QoS-1 and forwarder encapsulation.

MQTTSNGateway/src/MQTTGWConnectionHandler.cpp

@@ -90,10 +90,10 @@ void MQTTGWConnectionHandler::handlePingresp(Client* client, MQTTGWPacket* packe
 
 void MQTTGWConnectionHandler::handleDisconnect(Client* client, MQTTGWPacket* packet)
 {
-	MQTTSNPacket* snPacket = new MQTTSNPacket();
-	snPacket->setDISCONNECT(0);
-	client->disconnected();
-	client->getNetwork()->close();
-	Event* ev1 = new Event();
-	ev1->setClientSendEvent(client, snPacket);
+		MQTTSNPacket* snPacket = new MQTTSNPacket();
+		snPacket->setDISCONNECT(0);
+		client->disconnected();
+		client->getNetwork()->close();
+		Event* ev1 = new Event();
+		ev1->setClientSendEvent(client, snPacket);
 }

MQTTSNGateway/src/MQTTSNGWAggregater.cpp

@@ -40,7 +40,7 @@ void Aggregater::initialize(void)
 {
     char param[MQTTSNGW_PARAM_MAX];
 
-    if (_gateway->getParam("AggregateGateway", param) == 0 )
+    if (_gateway->getParam("AggregatingGateway", param) == 0 )
     {
         if (!strcasecmp(param, "YES") )
         {

MQTTSNGateway/src/MQTTSNGWClientList.cpp

@@ -24,6 +24,8 @@ extern Gateway* theGateway;
 /*=====================================
  Class ClientList
  =====================================*/
+const char* common_topic = "*";
+
 ClientList::ClientList()
 {
     _clientCnt = 0;
@@ -414,50 +416,58 @@ Client* ClientList::createClient(SensorNetAddress* addr, MQTTSNString* clientId,
 
 Client* ClientList::createPredefinedTopic( MQTTSNString* clientId, string topicName, uint16_t topicId, bool aggregate)
 {
-    Client* client = getClient(clientId);
+	if ( clientId->cstring == common_topic )
+	{
+		_gateway->getTopics()->add((const char*)topicName.c_str(), topicId);
+		return 0;
+	}
+	else
+	{
+		Client* client = getClient(clientId);
 
-    if ( _authorize && client == nullptr)
-    {
-        return 0;
-    }
+		if ( _authorize && client == nullptr )
+		{
+			return 0;
+		}
 
-    /*  anonimous clients */
-    if ( _clientCnt > MAX_CLIENTS )
-    {
-        return nullptr;  // full of clients
-    }
+		/*  anonimous clients */
+		if ( _clientCnt > MAX_CLIENTS )
+		{
+			return nullptr;  // full of clients
+		}
 
-    if ( client == nullptr )
-    {
-        /* creat a new client */
-        client = new Client();
-        client->setClientId(*clientId);
-        if ( aggregate )
-        {
-        	client->setAggregated();
-        }
-        _mutex.lock();
+		if ( client == nullptr )
+		{
+			/* creat a new client */
+			client = new Client();
+			client->setClientId(*clientId);
+			if ( aggregate )
+			{
+				client->setAggregated();
+			}
+			_mutex.lock();
 
-        /* add the list */
-        if ( _firstClient == nullptr )
-        {
-            _firstClient = client;
-            _endClient = client;
-        }
-        else
-        {
-            _endClient->_nextClient = client;
-            client->_prevClient = _endClient;
-            _endClient = client;
-        }
-        _clientCnt++;
-        _mutex.unlock();
-    }
+			/* add the list */
+			if ( _firstClient == nullptr )
+			{
+				_firstClient = client;
+				_endClient = client;
+			}
+			else
+			{
+				_endClient->_nextClient = client;
+				client->_prevClient = _endClient;
+				_endClient = client;
+			}
+			_clientCnt++;
+			_mutex.unlock();
+		}
 
-    // create Topic & Add it
-    client->getTopics()->add((const char*)topicName.c_str(), topicId);
-    client->_hasPredefTopic = true;
-    return client;
+		// create Topic & Add it
+		client->getTopics()->add((const char*)topicName.c_str(), topicId);
+		client->_hasPredefTopic = true;
+		return client;
+	}
 }
 
 uint16_t ClientList::getClientCount()

MQTTSNGateway/src/MQTTSNGWClientRecvTask.cpp

@@ -262,7 +262,7 @@ void ClientRecvTask::log(Client* client, MQTTSNPacket* packet, MQTTSNString* id)
 
 void ClientRecvTask::log(const char* clientId, MQTTSNPacket* packet)
 {
-    char pbuf[SIZE_OF_LOG_PACKET * 3];
+    char pbuf[ SIZE_OF_LOG_PACKET * 3 + 1];
     char msgId[6];
 
     switch (packet->getType())

MQTTSNGateway/src/MQTTSNGWClientSendTask.cpp

@@ -84,7 +84,7 @@ void ClientSendTask::run()
 
 void ClientSendTask::log(Client* client, MQTTSNPacket* packet)
 {
-	char pbuf[SIZE_OF_LOG_PACKET * 3];
+	char pbuf[SIZE_OF_LOG_PACKET * 3 + 1];
 	char msgId[6];
 	const char* clientId = client ? (const char*)client->getClientId() : UNKNOWNCL ;
 

MQTTSNGateway/src/MQTTSNGWPacket.cpp

@@ -423,12 +423,9 @@ char* MQTTSNPacket::print(char* pbuf)
 {
 	char* ptr = pbuf;
 	char** pptr = &pbuf;
-	int value = 0;
-
-	int i = MQTTSNPacket_decode(_buf, _bufLen, &value);
 	int size = _bufLen > SIZE_OF_LOG_PACKET ? SIZE_OF_LOG_PACKET : _bufLen;
 
-	for (; i < size; i++)
+	for (int i = 0; i < size; i++)
 	{
 		sprintf(*pptr, " %02X", *(_buf + i));
 		*pptr += 3;

MQTTSNGateway/src/MQTTSNGWPublishHandler.cpp

@@ -79,10 +79,18 @@ MQTTGWPacket* MQTTSNPublishHandler::handlePublish(Client* client, MQTTSNPacket*
 	else
 	{
 	    topic = client->getTopics()->getTopicById(&topicid);
+	    if ( !topic )
+	    {
+	    	topic = _gateway->getTopics()->getTopicById(&topicid);
+	    	if ( topic )
+	    	{
+	    		topic = client->getTopics()->add(topic->getTopicName()->c_str(), topic->getTopicId());
+	    	}
+	    }
 
 	    if( !topic && qos == 3 )
 	    {
-	        WRITELOG("%s Invali TopicId.%s %s\n", ERRMSG_HEADER, client->getClientId(), ERRMSG_FOOTER);
+	        WRITELOG("%s Invalid TopicId.%s %s\n", ERRMSG_HEADER, client->getClientId(), ERRMSG_FOOTER);
 	        return nullptr;
 	    }
 

MQTTSNGateway/src/MQTTSNGWSubscribeHandler.cpp

@@ -59,6 +59,8 @@ MQTTGWPacket* MQTTSNSubscribeHandler::handleSubscribe(Client* client, MQTTSNPack
     if ( topicFilter.type == MQTTSN_TOPIC_TYPE_PREDEFINED )
     {
         topic = client->getTopics()->getTopicById(&topicFilter);
+
+
         if ( topic )
         {
             topicId = topic->getTopicId();
@@ -67,7 +69,15 @@ MQTTGWPacket* MQTTSNSubscribeHandler::handleSubscribe(Client* client, MQTTSNPack
         }
         else
         {
-            goto RespExit;
+        	topic = _gateway->getTopics()->getTopicById(&topicFilter);
+        	if ( !topic )
+        	{
+				topic = client->getTopics()->add(topic->getTopicName()->c_str(), topic->getTopicId());
+			}
+        	else
+        	{
+        		goto RespExit;
+        	}
         }
     }
     else if (topicFilter.type == MQTTSN_TOPIC_TYPE_NORMAL)

MQTTSNGateway/src/MQTTSNGWVersion.h

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

MQTTSNGateway/src/MQTTSNGateway.cpp

@@ -37,6 +37,7 @@ Gateway::Gateway(void)
     _packetEventQue.setMaxSize(MAX_INFLIGHTMESSAGES * MAX_CLIENTS);
     _clientList = new ClientList();
     _adapterManager = new AdapterManager(this);
+    _topics = new Topics();
 }
 
 Gateway::~Gateway()
@@ -103,6 +104,11 @@ Gateway::~Gateway()
     {
         delete _clientList;
     }
+
+    if ( _topics )
+	{
+		delete _topics;
+	}
 }
 
 int Gateway::getParam(const char* parameter, char* value)
@@ -312,6 +318,11 @@ AdapterManager* Gateway::getAdapterManager(void)
     return _adapterManager;
 }
 
+Topics* Gateway::getTopics(void)
+{
+    return _topics;
+}
+
 bool Gateway::hasSecureConnection(void)
 {
 	return (  _params.certKey

MQTTSNGateway/src/MQTTSNGateway.h

@@ -191,6 +191,7 @@ public:
 	AdapterManager* getAdapterManager(void);
 	int getParam(const char* parameter, char* value);
 	bool hasSecureConnection(void);
+	Topics* getTopics(void);
 
 private:
 	GatewayParams  _params;
@@ -201,6 +202,7 @@ private:
 	LightIndicator _lightIndicator;
 	SensorNetwork  _sensorNetwork;
 	AdapterManager* _adapterManager {nullptr};
+	Topics* _topics;
 };
 
 }