BugFix: Network can not handle EPIPE error. Issue#35 #36

Signed-off-by: tomoaki <tomoaki@tomy-tech.com>
This commit is contained in:
tomoaki
2016-10-07 18:55:03 +09:00
parent d834a1a65b
commit 9de0448130
13 changed files with 246 additions and 252 deletions

View File

@@ -22,7 +22,8 @@ ClientAuthentication=NO
#RootCAfile=/etc/ssl/certs/ca-certificates.crt
#RootCApath=/etc/ssl/certs/
#CertsDirectory=/usr/share/GW/IoTcerts/
#CertsFile=/path/to/certKey.pem
#PrivateKey=/path/to/privateKey.pem
GatewayID=1
GatewayName=PahoGateway-01

View File

@@ -184,9 +184,10 @@ int MQTTGWPacket::recv(Network* network)
unsigned char c;
/* read First Byte of Packet */
if (network->recv((unsigned char*)&_header.byte, 1) == -1)
int rc = network->recv((unsigned char*)&_header.byte, 1);
if ( rc <= 0)
{
return -1;
return rc;
}
/* read RemainingLength */
do

View File

@@ -59,8 +59,10 @@ void BrokerRecvTask::run(void)
while (true)
{
_light->blueLight(false);
if (CHK_SIGINT)
{
WRITELOG("%s BrokerRecvTask stopped.\n", currentDateTime());
return;
}
timeout.tv_sec = 0;
@@ -72,7 +74,6 @@ void BrokerRecvTask::run(void)
/* Prepare sockets list to read */
Client* client = _gateway->getClientList()->getClient();
_light->blueLight(false);
while (client > 0)
{
@@ -115,7 +116,8 @@ void BrokerRecvTask::run(void)
if ( log(client, packet) == -1 )
{
continue;
delete packet;
goto nextClient;
}
/* post a BrokerRecvEvent */
@@ -125,13 +127,7 @@ void BrokerRecvTask::run(void)
}
else
{
_light->blueLight(false);
if ( rc == 0 )
{
delete packet;
continue;
}
else if (rc == -1)
if (rc == -1)
{
WRITELOG("%s BrokerRecvTask can't receive a packet from the broker errno=%d %s%s\n", ERRMSG_HEADER, errno, client->getClientId(), ERRMSG_FOOTER);
}
@@ -141,14 +137,14 @@ void BrokerRecvTask::run(void)
}
else if ( rc == -3 )
{
WRITELOG("%s BrokerRecvTask can't create the packet %s%s\n", ERRMSG_HEADER, client->getClientId(), ERRMSG_FOOTER);
WRITELOG("%s BrokerRecvTask can't get memories for the packet %s%s\n", ERRMSG_HEADER, client->getClientId(), ERRMSG_FOOTER);
}
delete packet;
/* disconnect the client */
if ( rc == -1 || rc == -2 )
if ( (rc == -1 || rc == -2) && client->isActive() )
{
/* disconnect the client */
packet = new MQTTGWPacket();
packet->setHeader(DISCONNECT);
ev = new Event();
@@ -158,16 +154,11 @@ void BrokerRecvTask::run(void)
}
}
}
nextClient:
client = client->getNextClient();
}
_light->blueLight(false);
}
}
else
{
_light->greenLight(false);
}
maxSock = 0;
}
}
@@ -202,6 +193,7 @@ int BrokerRecvTask::log(Client* client, MQTTGWPacket* packet)
WRITELOG(FORMAT_Y_Y_W, currentDateTime(), packet->getName(), LEFTARROW, client->getClientId(), packet->print(pbuf));
break;
default:
WRITELOG("Type=%x\n", packet->getType());
rc = -1;
break;
}

View File

@@ -68,6 +68,7 @@ void BrokerSendTask::run()
if ( ev->getEventType() == EtStop )
{
WRITELOG("%s BrokerSendTask stopped.\n", currentDateTime());
delete ev;
return;
}
@@ -77,34 +78,23 @@ void BrokerSendTask::run()
client = ev->getClient();
packet = ev->getMQTTGWPacket();
if ( packet->getType() == CONNECT && client->getNetwork()->isValid() )
{
client->getNetwork()->close();
}
if ( !client->getNetwork()->isValid() )
{
/* connect to the broker and send a packet */
char* portNo = _gwparams->port;
const char* cert = 0;
const char* keyFile = 0;
string certFile;
string privateKeyFile;
if (client->isSecureNetwork())
{
portNo = _gwparams->portSecure;
if ( _gwparams->certDirectory )
{
certFile = _gwparams->certDirectory;
certFile += client->getClientId();
certFile += ".crt";
cert = certFile.c_str();
privateKeyFile = _gwparams->certDirectory;
privateKeyFile += client->getClientId();
privateKeyFile += ".key";
keyFile = privateKeyFile.c_str();
}
rc = client->getNetwork()->connect(_gwparams->brokerName, _gwparams->portSecure, _gwparams->rootCApath, _gwparams->rootCAfile, cert, keyFile);
rc = client->getNetwork()->connect(_gwparams->brokerName, _gwparams->portSecure, _gwparams->rootCApath,
_gwparams->rootCAfile, _gwparams->certKey, _gwparams->privateKey);
}
else
{
rc = client->getNetwork()->connect(_gwparams->brokerName, portNo);
rc = client->getNetwork()->connect(_gwparams->brokerName, _gwparams->port);
}
if ( !rc )
@@ -133,13 +123,14 @@ void BrokerSendTask::run()
{
WRITELOG("%s BrokerSendTask can't send a packet to the broker errno=%d %s%s\n",
ERRMSG_HEADER, rc == -1 ? errno : 0, client->getClientId(), ERRMSG_FOOTER);
client->getNetwork()->close();
/* Disconnect the client */
packet = new MQTTGWPacket();
packet->setHeader(DISCONNECT);
ev = new Event();
ev->setBrokerRecvEvent(client, packet);
_gateway->getPacketEventQue()->post(ev);
Event* ev1 = new Event();
ev1->setBrokerRecvEvent(client, packet);
_gateway->getPacketEventQue()->post(ev1);
}
_light->blueLight(false);

View File

@@ -61,6 +61,7 @@ void ClientRecvTask::run()
if (CHK_SIGINT)
{
WRITELOG("%s ClientRecvTask stopped.\n", currentDateTime());
delete packet;
return;
}
@@ -109,7 +110,7 @@ void ClientRecvTask::run()
/* create a client */
client = _gateway->getClientList()->createClient(_sensorNetwork->getSenderAddress(), &data.clientID, false, false);
log(client, packet, &data.clientID);
if (!client)
{
WRITELOG("%s Client was rejected. CONNECT message has been discarded.%s\n", ERRMSG_HEADER, ERRMSG_FOOTER);
@@ -117,8 +118,6 @@ void ClientRecvTask::run()
continue;
}
log(client, packet);
/* set sensorNetAddress & post Event */
client->setClientAddress(_sensorNetwork->getSenderAddress());
ev = new Event();
@@ -143,11 +142,27 @@ void ClientRecvTask::run()
}
}
void ClientRecvTask::log(Client* client, MQTTSNPacket* packet)
void ClientRecvTask::log(Client* client, MQTTSNPacket* packet, MQTTSNString* id)
{
char pbuf[SIZE_OF_LOG_PACKET * 3];
const char* clientId;
char cstr[MAX_CLIENTID_LENGTH + 1];
char msgId[6];
const char* clientId = client ? (const char*)client->getClientId() : NONACTCLT ;
if ( id )
{
memset((void*)cstr, 0, id->lenstring.len);
strncpy(cstr, id->lenstring.data, id->lenstring.len) ;
clientId = cstr;
}
else if ( client )
{
clientId = client->getClientId();
}
else
{
clientId = UNKNOWNCL;
}
switch (packet->getType())
{

View File

@@ -35,7 +35,7 @@ public:
void run();
private:
void log(Client*, MQTTSNPacket*);
void log(Client*, MQTTSNPacket*, MQTTSNString* id = 0);
Gateway* _gateway;
SensorNetwork* _sensorNetwork;

View File

@@ -47,6 +47,7 @@ void ClientSendTask::run()
if (ev->getEventType() == EtStop)
{
WRITELOG("%s ClientSendTask stopped.\n", currentDateTime());
delete ev;
break;
}
@@ -76,7 +77,7 @@ void ClientSendTask::log(Client* client, MQTTSNPacket* packet)
{
char pbuf[SIZE_OF_LOG_PACKET * 3];
char msgId[6];
const char* clientId = client ? (const char*)client->getClientId() : NONACTCLT ;
const char* clientId = client ? (const char*)client->getClientId() : UNKNOWNCL ;
switch (packet->getType())
{

View File

@@ -26,6 +26,12 @@ namespace MQTTSNGW
#define CONFIG_FILE "gateway.conf"
#define CLIENT_LIST "clients.conf"
/*==========================================================
* Gateway default parameters
===========================================================*/
#define DEFAULT_KEEP_ALIVE_TIME (900) // 900 secs = 15 mins
#define DEFAULT_MQTT_VERSION (4) // Defualt MQTT version
/*=================================
* MQTT-SN Parametrs
==================================*/

View File

@@ -32,7 +32,7 @@ using namespace std;
using namespace MQTTSNGW;
#define EVENT_QUE_TIME_OUT 2000 // 2000 msecs
char* currentDateTime(void);
/*=====================================
Class PacketHandleTask
=====================================*/
@@ -99,24 +99,13 @@ void PacketHandleTask::run()
if (ev->getEventType() == EtStop)
{
WRITELOG("%s PacketHandleTask stopped.\n", currentDateTime());
delete ev;
return;
}
if (ev->getEventType() == EtTimeout)
{
/*------ Is Client Lost ? ---------*/
/*
client = _gateway->getClientList()->getClient();
while (client > 0)
{
if ( client->checkTimeover() )
{
client->disconnected();
}
client = client->getNextClient();
}
*/
/*------ Check Keep Alive Timer & send Advertise ------*/
if (_advertiseTimer.isTimeup())
{

View File

@@ -41,7 +41,8 @@ Gateway::Gateway()
_params.portSecure = 0;
_params.rootCApath = 0;
_params.rootCAfile = 0;
_params.certDirectory = 0;
_params.certKey = 0;
_params.privateKey = 0;
_params.clientListName = 0;
_params.configName = 0;
_packetEventQue.setMaxSize(MAX_INFLIGHTMESSAGES * MAX_CLIENTS);
@@ -73,9 +74,13 @@ Gateway::~Gateway()
{
free(_params.portSecure);
}
if ( _params.certDirectory )
if ( _params.certKey )
{
free(_params.certDirectory);
free(_params.certKey);
}
if ( _params.privateKey )
{
free(_params.privateKey);
}
if ( _params.rootCApath )
{
@@ -115,9 +120,13 @@ void Gateway::initialize(int argc, char** argv)
_params.portSecure = strdup(param);
}
if (getParam("CertsDirectory", param) == 0)
if (getParam("CertKey", param) == 0)
{
_params.certDirectory = strdup(param);
_params.certKey = strdup(param);
}
if (getParam("PrivateKey", param) == 0)
{
_params.privateKey = strdup(param);
}
if (getParam("RootCApath", param) == 0)
{
@@ -221,7 +230,8 @@ void Gateway::run(void)
WRITELOG(" Broker: %s : %s, %s\n", _params.brokerName, _params.port, _params.portSecure);
WRITELOG(" RootCApath: %s\n", _params.rootCApath);
WRITELOG(" RootCAfile: %s\n", _params.rootCAfile);
WRITELOG(" ClientCerts: %s\n", _params.certDirectory);
WRITELOG(" CertKey: %s\n", _params.certKey);
WRITELOG(" PrivateKey: %s\n", _params.privateKey);
MultiTaskProcess::run();
@@ -239,7 +249,7 @@ void Gateway::run(void)
/* wait until all threads stop */
MultiTaskProcess::waitStop();
WRITELOG("%s MQTT-SN Gateway stoped\n", currentDateTime());
WRITELOG("\n%s MQTT-SN Gateway stoped\n\n", currentDateTime());
_lightIndicator.allLightOff();
}

View File

@@ -22,16 +22,10 @@
namespace MQTTSNGW
{
/*==========================================================
* Gateway default parameters
===========================================================*/
#define DEFAULT_KEEP_ALIVE_TIME (900) // 900 secs = 15 mins
#define DEFAULT_MQTT_VERSION (4) // Defualt MQTT version
/*=================================
* Starting prompt
==================================*/
#define GATEWAY_VERSION " * Version: 0.9.0"
#define GATEWAY_VERSION " * Version: 0.9.1"
#define PAHO_COPYRIGHT0 " * MQTT-SN Transparent Gateway"
#define PAHO_COPYRIGHT1 " * Part of Project Paho in Eclipse"
@@ -49,7 +43,7 @@ namespace MQTTSNGW
===========================================================*/
#define CLIENT "Client"
#define CLIENTS "Clients"
#define NONACTCLT "Non Active Client !"
#define UNKNOWNCL "Unknown Client !"
#define LEFTARROW "<---"
#define RIGHTARROW "--->"
@@ -164,7 +158,8 @@ typedef struct
char* portSecure;
char* rootCApath;
char* rootCAfile;
char* certDirectory;
char* certKey;
char* privateKey;
}GatewayParams;
/*=====================================

View File

@@ -158,7 +158,7 @@ bool TCPStack::connect(const char* host, const char* service)
{
if (isValid())
{
return false;
return true;
}
addrinfo hints;
memset(&hints, 0, sizeof(addrinfo));
@@ -192,7 +192,7 @@ bool TCPStack::connect(const char* host, const char* service)
if (::connect(sockfd, _addrinfo->ai_addr, _addrinfo->ai_addrlen) < 0)
{
//perror("TCPStack connect");
DEBUGLOG("Can not connect the socket. Check the PortNo! \n");
::close(sockfd);
return false;
}
@@ -233,6 +233,7 @@ int TCPStack::getSock()
=======================================*/
int Network::_numOfInstance = 0;
SSL_CTX* Network::_ctx = 0;
SSL_SESSION* Network::_session = 0;
Network::Network(bool secure) :
TCPStack()
@@ -240,57 +241,49 @@ Network::Network(bool secure) :
_ssl = 0;
_secureFlg = secure;
_busy = false;
_session = 0;
_sslValid = false;
}
Network::~Network()
{
if (_ssl)
{
SSL_shutdown(_ssl);
SSL_free(_ssl);
_numOfInstance--;
}
if (_session )
{
SSL_SESSION_free(_session);
}
if (_ctx && _numOfInstance == 0)
{
SSL_CTX_free(_ctx);
_ctx = 0;
ERR_free_strings();
}
close();
}
bool Network::connect(const char* host, const char* port)
{
bool rc = false;
_mutex.lock();
if (_secureFlg)
{
return false;
goto exit;
}
if (getSock() == 0)
{
if (!TCPStack::connect(host, port))
{
return false;
goto exit;
}
}
return true;
rc = true;
exit:
_mutex.unlock();
return rc;
}
bool Network::connect(const char* host, const char* port, const char* caPath, const char* caFile, const char* cert, const char* prvkey)
bool Network::connect(const char* host, const char* port, const char* caPath, const char* caFile, const char* certkey, const char* prvkey)
{
char errmsg[256];
char peer_CN[256];
int rc = 0;
bool rc;
_mutex.lock();
try
{
if (!_secureFlg)
{
WRITELOG("TLS is not required.\n");
return false;
throw;
}
if (_ctx == 0)
@@ -302,39 +295,42 @@ bool Network::connect(const char* host, const char* port, const char* caPath, co
{
ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
WRITELOG("SSL_CTX_new() %s\n", errmsg);
return false;
throw;
}
if (!SSL_CTX_load_verify_locations(_ctx, caFile, caPath))
{
ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
WRITELOG("SSL_CTX_load_verify_locations() %s\n", errmsg);
return false;
throw;
}
if ( certkey )
{
if ( SSL_CTX_use_certificate_file(_ctx, certkey, SSL_FILETYPE_PEM) != 1 )
{
ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
WRITELOG("SSL_CTX_use_certificate_file() %s %s\n", certkey, errmsg);
throw;
}
}
if ( prvkey )
{
if ( SSL_CTX_use_PrivateKey_file(_ctx, prvkey, SSL_FILETYPE_PEM) != 1 )
{
ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
WRITELOG("SSL_use_PrivateKey_file() %s %s\n", prvkey, errmsg);
throw;
}
}
}
if (!_sslValid)
if (! TCPStack::isValid())
{
if ( !TCPStack::connect(host, port) )
{
return false;
throw;
}
/*
if ( _ssl )
{
if (!SSL_set_fd(_ssl, getSock()))
{
ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
WRITELOG("SSL_set_fd() %s\n", errmsg);
SSL_free(_ssl);
}
else
{
_sslValid = true;
return true;
}
}
*/
}
_ssl = SSL_new(_ctx);
@@ -342,42 +338,7 @@ bool Network::connect(const char* host, const char* port, const char* caPath, co
{
ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
WRITELOG("SSL_new() %s\n", errmsg);
return false;
}
if (_session)
{
rc = SSL_set_session(_ssl, _session);
}
if (!SSL_set_fd(_ssl, getSock()))
{
ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
WRITELOG("SSL_set_fd() %s\n", errmsg);
SSL_free(_ssl);
}
if ( cert )
{
if ( SSL_use_certificate_file(_ssl, cert, SSL_FILETYPE_PEM) != 1 )
{
ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
WRITELOG("SSL_use_certificate_file() %s %s\n", cert, errmsg);
SSL_free(_ssl);
_ssl = 0;
return false;
}
}
if ( prvkey )
{
if ( SSL_use_PrivateKey_file(_ssl, prvkey, SSL_FILETYPE_PEM) != 1 )
{
ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
WRITELOG("SSL_use_PrivateKey_file() %s %s\n", prvkey, errmsg);
SSL_free(_ssl);
_ssl = 0;
return false;
}
throw;
}
if (!SSL_set_fd(_ssl, TCPStack::getSock()))
@@ -386,7 +347,12 @@ bool Network::connect(const char* host, const char* port, const char* caPath, co
WRITELOG("SSL_set_fd() %s\n", errmsg);
SSL_free(_ssl);
_ssl = 0;
return false;
throw;
}
if (_session)
{
SSL_set_session(_ssl, _session);
}
if (SSL_connect(_ssl) != 1)
@@ -395,15 +361,16 @@ bool Network::connect(const char* host, const char* port, const char* caPath, co
WRITELOG("SSL_connect() %s\n", errmsg);
SSL_free(_ssl);
_ssl = 0;
return false;
throw;
}
if ( (rc = SSL_get_verify_result(_ssl)) != X509_V_OK)
int result;
if ( (result = SSL_get_verify_result(_ssl)) != X509_V_OK)
{
WRITELOG("SSL_get_verify_result() error: %s.\n", X509_verify_cert_error_string(rc));
WRITELOG("SSL_get_verify_result() error: %s.\n", X509_verify_cert_error_string(result));
SSL_free(_ssl);
_ssl = 0;
return false;
throw;
}
X509* peer = SSL_get_peer_certificate(_ssl);
@@ -419,16 +386,23 @@ bool Network::connect(const char* host, const char* port, const char* caPath, co
WRITELOG("SSL_get_peer_certificate() error: Broker %s dosen't match the host name %s\n", peer_CN, host);
SSL_free(_ssl);
_ssl = 0;
return false;
throw;
}
if (_session == 0)
{
//_session = SSL_get1_session(_ssl);
_session = SSL_get1_session(_ssl);
}
_numOfInstance++;
_sslValid = true;
return true;
rc = true;
}
catch (...)
{
rc = false;
}
_mutex.unlock();
return rc;
}
int Network::send(const uint8_t* buf, uint16_t length)
@@ -446,6 +420,12 @@ int Network::send(const uint8_t* buf, uint16_t length)
else
{
_mutex.lock();
if ( !_ssl )
{
_mutex.unlock();
return -1;
}
_busy = true;
while (true)
@@ -515,8 +495,14 @@ int Network::recv(uint8_t* buf, uint16_t len)
return 0;
}
_mutex.lock();
_busy = true;
if ( !_ssl )
{
_mutex.unlock();
return 0;
}
_busy = true;
loop: do
{
readBlockedOnWrite = false;
@@ -535,7 +521,7 @@ int Network::recv(uint8_t* buf, uint16_t len)
SSL_shutdown(_ssl);
_ssl = 0;
_numOfInstance--;
TCPStack::close();
//TCPStack::close();
_busy = false;
_mutex.unlock();
return -1;
@@ -585,16 +571,37 @@ int Network::recv(uint8_t* buf, uint16_t len)
void Network::close(void)
{
_mutex.lock();
if (_secureFlg)
{
_sslValid = false;
if (_ssl)
{
SSL_shutdown(_ssl);
SSL_free(_ssl);
_numOfInstance--;
_ssl = 0;
_sslValid = false;
_busy = false;
}
if (_session && _numOfInstance == 0)
{
SSL_SESSION_free(_session);
_session = 0;
}
if (_ctx && _numOfInstance == 0)
{
SSL_CTX_free(_ctx);
_ctx = 0;
ERR_free_strings();
}
}
TCPStack::close();
_mutex.unlock();
}
bool Network::isValid()
{
if ( TCPStack::isValid() )
{
if (_secureFlg)
{
@@ -605,24 +612,12 @@ bool Network::isValid()
}
else
{
return TCPStack::isValid();
return true;
}
}
return false;
}
void Network::disconnect()
{
if (_ssl)
{
SSL_shutdown(_ssl);
_ssl = 0;
}
_sslValid = false;
_busy = false;
TCPStack::close();
}
int Network::getSock()
{
return TCPStack::getSock();

View File

@@ -71,9 +71,8 @@ public:
Network(bool secure);
virtual ~Network();
bool connect(const char* host, const char* port, const char* caPath, const char* caFile, const char* sert, const char* prvkey);
bool connect(const char* host, const char* port, const char* caPath, const char* caFile, const char* cert, const char* prvkey);
bool connect(const char* host, const char* port);
void disconnect(void);
void close(void);
int send(const uint8_t* buf, uint16_t length);
int recv(uint8_t* buf, uint16_t len);
@@ -84,9 +83,8 @@ public:
private:
static SSL_CTX* _ctx;
static SSL_SESSION* _session;
static int _numOfInstance;
SSL_SESSION* _session;
SSL* _ssl;
bool _secureFlg;
Mutex _mutex;