MQTTSNGWClientSendTask.cpp

00001 /**************************************************************************************
00002  * Copyright (c) 2016, Tomoaki Yamaguchi
00003  *
00004  * All rights reserved. This program and the accompanying materials
00005  * are made available under the terms of the Eclipse Public License v1.0
00006  * and Eclipse Distribution License v1.0 which accompany this distribution.
00007  *
00008  * The Eclipse Public License is available at
00009  *    http://www.eclipse.org/legal/epl-v10.html
00010  * and the Eclipse Distribution License is available at
00011  *   http://www.eclipse.org/org/documents/edl-v10.php.
00012  *
00013  * Contributors:
00014  *    Tomoaki Yamaguchi - initial API and implementation and/or initial documentation
00015  **************************************************************************************/
00016 #include "MQTTSNGWClientSendTask.h"
00017 #include "MQTTSNGWPacket.h"
00018 #include "MQTTSNGateway.h"
00019 #include "MQTTSNGWEncapsulatedPacket.h"
00020 #include "MQTTSNGWQoSm1Proxy.h"
00021 
00022 using namespace MQTTSNGW;
00023 using namespace std;
00024 char* currentDateTime(void);
00025 /*=====================================
00026  Class ClientSendTask
00027  =====================================*/
00028 ClientSendTask::ClientSendTask(Gateway* gateway)
00029 {
00030     _gateway = gateway;
00031     _gateway->attach((Thread*)this);
00032     _sensorNetwork = _gateway->getSensorNetwork();
00033 }
00034 
00035 ClientSendTask::~ClientSendTask()
00036 {
00037 
00038 }
00039 
00040 void ClientSendTask::run()
00041 {
00042     Client* client = nullptr;
00043     MQTTSNPacket* packet = nullptr;
00044     AdapterManager* adpMgr = _gateway->getAdapterManager();
00045     int rc = 0;
00046 
00047     while (true)
00048     {
00049         Event* ev = _gateway->getClientSendQue()->wait();
00050 
00051         if (ev->getEventType() == EtStop)
00052         {
00053             WRITELOG("%s ClientSendTask   stopped.\n", currentDateTime());
00054             delete ev;
00055             break;
00056         }
00057         if (ev->getEventType() == EtClientSend)
00058         {
00059             client = ev->getClient();
00060             packet = ev->getMQTTSNPacket();
00061             rc = adpMgr->unicastToClient(client, packet, this);
00062         }
00063         else if (ev->getEventType() == EtBroadcast)
00064         {
00065             packet = ev->getMQTTSNPacket();
00066             log(client, packet);
00067             rc = packet->broadcast(_sensorNetwork);
00068         }
00069         else if (ev->getEventType() == EtSensornetSend)
00070         {
00071             packet = ev->getMQTTSNPacket();
00072             log(client, packet);
00073             rc = packet->unicast(_sensorNetwork, ev->getSensorNetAddress());
00074         }
00075 
00076         if ( rc < 0 )
00077         {
00078             WRITELOG("%s ClientSendTask can't send a packet to the client %s%s.\n",
00079                 ERRMSG_HEADER, (client ? (const char*)client->getClientId() : UNKNOWNCL ), ERRMSG_FOOTER);
00080         }
00081         delete ev;
00082     }
00083 }
00084 
00085 void ClientSendTask::log(Client* client, MQTTSNPacket* packet)
00086 {
00087     char pbuf[SIZE_OF_LOG_PACKET * 3 + 1];
00088     char msgId[6];
00089     const char* clientId = client ? (const char*)client->getClientId() : UNKNOWNCL ;
00090 
00091     switch (packet->getType())
00092     {
00093     case MQTTSN_ADVERTISE:
00094     case MQTTSN_GWINFO:
00095         WRITELOG(FORMAT_Y_W_G, currentDateTime(), packet->getName(), RIGHTARROW, CLIENTS, packet->print(pbuf));
00096         break;
00097     case MQTTSN_CONNACK:
00098     case MQTTSN_DISCONNECT:
00099     case MQTTSN_WILLTOPICREQ:
00100     case MQTTSN_WILLMSGREQ:
00101     case MQTTSN_WILLTOPICRESP:
00102     case MQTTSN_WILLMSGRESP:
00103     case MQTTSN_PINGRESP:
00104         WRITELOG(FORMAT_Y_W_G, currentDateTime(), packet->getName(), RIGHTARROW, clientId, packet->print(pbuf));
00105         break;
00106     case MQTTSN_REGISTER:
00107     case MQTTSN_PUBLISH:
00108         WRITELOG(FORMAT_W_MSGID_W_G, currentDateTime(), packet->getName(), packet->getMsgId(msgId), RIGHTARROW, clientId,   packet->print(pbuf));
00109         break;
00110     case MQTTSN_REGACK:
00111     case MQTTSN_PUBACK:
00112     case MQTTSN_PUBREC:
00113     case MQTTSN_PUBREL:
00114     case MQTTSN_PUBCOMP:
00115     case MQTTSN_SUBACK:
00116     case MQTTSN_UNSUBACK:
00117         WRITELOG(FORMAT_W_MSGID_W_G, currentDateTime(), packet->getName(), packet->getMsgId(msgId), RIGHTARROW, clientId,   packet->print(pbuf));
00118         break;
00119     default:
00120         break;
00121     }
00122 }
00123