GitHub/paho.mqtt-sn.embedded-c
1
0
Fork 0
mirror of https://github.com/eclipse/paho.mqtt-sn.embedded-c.git synced 2025-12-16 08:56:51 +01:00
Code Issues Projects Releases Wiki Activity

This branch for debugging DTLS #90, #150, #195, #227

Browse Source 
The purpose of this branch is to share work in process.
Change sellect() of UDP to poll()
Rewrite UDP6 for DTLS6

Known bug: can't reconnect DTLS

Signed-off-by: tomoaki <tomoaki@tomy-tech.com>
This commit is contained in:
tomoaki
2021-06-11 10:54:20 +09:00
parent 83c30d662f
commit 5fb4312aad
52 changed files with 5101 additions and 1790 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
MQTTSNGateway/src/CMakeLists.txt
View File

@@ -35,6 +35,9 @@ IF(NOT DEFINED SENSORNET)
ENDIF()
MESSAGE(STATUS "SENSORNET: " ${SENSORNET})
ADD_DEFINITIONS(${DEFS})
MESSAGE(STATUS "Definitions: " ${DEFS})
ADD_LIBRARY(mqtt-sngateway_common
MQTTGWConnectionHandler.cpp
MQTTGWPacket.cpp
@@ -74,8 +77,11 @@ ADD_LIBRARY(mqtt-sngateway_common
${OS}/Threading.h
)
# linux
link_directories("/usr/local/lib")
link_directories("/usr/local/opt/openssl/lib")
# Mac
link_directories("/usr/local/opt/openssl")
TARGET_INCLUDE_DIRECTORIES(mqtt-sngateway_common
@@ -84,7 +90,7 @@ TARGET_INCLUDE_DIRECTORIES(mqtt-sngateway_common
${OS}
${OS}/${SENSORNET}
../../MQTTSNPacket/src
/usr/local/opt/openssl/include
/usr/local/include
)
IF(SENSORNET MATCHES "rfcomm")

1
MQTTSNGateway/src/MQTTSNGWBrokerRecvTask.cpp
View File

@@ -17,6 +17,7 @@
#include "MQTTSNGWBrokerRecvTask.h"
#include "MQTTSNGWClient.h"
#include "MQTTSNGWClientList.h"
#include "MQTTSNGateway.h"
#include <unistd.h>
using namespace std;

60
MQTTSNGateway/src/MQTTSNGWClientList.cpp
View File

@@ -58,6 +58,7 @@ ClientList::~ClientList()
void ClientList::initialize(bool aggregate)
{
_maxClients = _gateway->getGWParams()->maxClients;
_clientsPool->allocate(_gateway->getGWParams()->maxClients);
if (_gateway->getGWParams()->clientAuthentication)
@@ -380,6 +381,7 @@ Client* ClientList::createClient(SensorNetAddress* addr, MQTTSNString* clientId,
{
client->setQoSm1();
}
client->getNetwork()->setSecure(secure);
_mutex.lock();
@@ -402,16 +404,18 @@ Client* ClientList::createClient(SensorNetAddress* addr, MQTTSNString* clientId,
Client* ClientList::createPredefinedTopic(MQTTSNString* clientId, string topicName, uint16_t topicId, bool aggregate)
{
Client *client = nullptr;
if (topicId == 0)
{
WRITELOG("Invalid TopicId. Predefined Topic %s, TopicId is 0. \n", topicName.c_str());
return nullptr;
goto exit;
}
if (strcmp(clientId->cstring, common_topic) == 0)
{
_gateway->getTopics()->add((const char*) topicName.c_str(), topicId);
return nullptr;
goto exit;
}
else
{
@@ -419,47 +423,19 @@ Client* ClientList::createPredefinedTopic(MQTTSNString* clientId, string topicNa
if (_authorize && client == nullptr)
{
return nullptr;
goto exit;
}
/* anonimous clients */
if (_clientCnt > MAX_CLIENTS)
client = createClient(NULL, clientId, aggregate);
if (client)
{
return nullptr; // full of clients
// create Topic & Add it
client->getTopics()->add((const char*) topicName.c_str(), topicId);
client->_hasPredefTopic = true;
}
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();
}
// create Topic & Add it
client->getTopics()->add((const char*) topicName.c_str(), topicId);
client->_hasPredefTopic = true;
return client;
}
exit:
return client;
}
uint16_t ClientList::getClientCount()
@@ -517,15 +493,17 @@ void ClientsPool::allocate(int maxClients)
Client* ClientsPool::getClient(void)
{
Client *cl = nullptr;
while (_firstClient != nullptr)
{
Client* cl = _firstClient;
cl = _firstClient;
_firstClient = cl->_nextClient;
cl->_nextClient = nullptr;
_clientCnt--;
return cl;
break;
}
return nullptr;
return cl;
}
void ClientsPool::setClient(Client* client)

1
MQTTSNGateway/src/MQTTSNGWClientList.h
View File

@@ -83,6 +83,7 @@ private:
Client* _endClient;
Mutex _mutex;
uint16_t _clientCnt;
uint16_t _maxClients;
bool _authorize { false };
};

14
MQTTSNGateway/src/MQTTSNGWDefines.h
View File

@@ -37,7 +37,7 @@ namespace MQTTSNGW
/*=================================
* MQTT-SN Parametrs
==================================*/
#define MAX_CLIENTS (100) // Number of Clients can be handled.
#define MAX_CLIENTS (100) // Default number of Clients can be handled.
#define MAX_CLIENTID_LENGTH (64) // Max length of clientID
#define MAX_INFLIGHTMESSAGES (10) // Number of inflight messages
#define MAX_MESSAGEID_TABLE_SIZE (500) // Number of MessageIdTable size
@@ -60,13 +60,19 @@ typedef unsigned int uint32_t;
/*=================================
* Log controls
==================================*/
//#define DEBUG // print out log for debug
//#define DEBUG_NWSTACK // print out SensorNetwork log
#ifdef DEBUG
//#define DEBUG_MQTTSN // print out log for debug
//#define DEBUG_NW // print out SensorNetwork log
#ifdef DEBUG_MQTTSN
#define DEBUGLOG(...) printf(__VA_ARGS__)
#else
#define DEBUGLOG(...)
#endif
#ifdef DEBUG_NW
#define D_NWSTACK(...) printf(__VA_ARGS__)
#else
#define D_NWSTACK(...)
#endif
}
#endif /* MQTTSNGWDEFINES_H_ */

4
MQTTSNGateway/src/MQTTSNGWPacket.cpp
View File

@@ -100,10 +100,6 @@ int MQTTSNPacket::recv(SensorNetwork* network)
{
len = desirialize(buf, len);
}
else
{
len = 0;
}
return len;
}

1
MQTTSNGateway/src/MQTTSNGWPacket.h
View File

@@ -22,6 +22,7 @@
namespace MQTTSNGW
{
class SensorNetwork;
class MQTTSNPacket
{

1
MQTTSNGateway/src/MQTTSNGWPacketHandleTask.cpp
View File

@@ -38,6 +38,7 @@ using namespace MQTTSNGW;
#define EVENT_QUE_TIME_OUT 2000 // 2000 msecs
char* currentDateTime(void);
/*=====================================
Class PacketHandleTask
=====================================*/

64
MQTTSNGateway/src/MQTTSNGWProcess.cpp
View File

@@ -16,6 +16,7 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <string>
#include <stdarg.h>
#include <signal.h>
#include <Timer.h>
@@ -156,6 +157,8 @@ int Process::getParam(const char* parameter, char* value)
{
char str[MQTTSNGW_PARAM_MAX];
char param[MQTTSNGW_PARAM_MAX];
memset(str, 0, sizeof(str));
memset(param, 0, sizeof(param));
FILE *fp;
int i = 0, j = 0;
@@ -166,40 +169,55 @@ int Process::getParam(const char* parameter, char* value)
throw Exception("Config file not found:\n\nUsage: Command -f path/config_file_name\n", 0);
}
int paramlen = strlen(parameter);
while (true)
{
int pos = 0;
int len = 0;
if (fgets(str, MQTTSNGW_PARAM_MAX - 1, fp) == NULL)
{
fclose(fp);
return -3;
}
if (!strncmp(str, parameter, strlen(parameter)))
if (str[0] == '#' || str[0] == '\n')
{
while (str[i++] != '=')
{
;
}
while (str[i] != '\n')
{
param[j++] = str[i++];
}
param[j] = '\0';
continue;
}
for (i = strlen(param) - 1; i >= 0 && isspace(param[i]); i--)
;
param[i + 1] = '\0';
for (i = 0; isspace(param[i]); i++)
;
if (i > 0)
len = strlen(str);
for (pos = 0; i < len; pos++)
{
if (str[pos] == '=')
{
j = 0;
while (param[i])
param[j++] = param[i++];
param[j] = '\0';
break;
}
}
if (pos == paramlen)
{
if (strncmp(str, parameter, paramlen) == 0)
{
strcpy(param, str + pos + 1);
param[len - pos - 2] = '\0';
for (i = strlen(param) - 1; i >= 0 && isspace(param[i]); i--)
;
param[i + 1] = '\0';
for (i = 0; isspace(param[i]); i++)
;
if (i > 0)
{
j = 0;
while (param[i])
param[j++] = param[i++];
param[j] = '\0';
}
strcpy(value, param);
fclose(fp);
return 0;
}
strcpy(value, param);
fclose(fp);
return 0;
}
}
fclose(fp);

60
MQTTSNGateway/src/MQTTSNGateway.cpp
View File

@@ -32,9 +32,9 @@ MQTTSNGW::Gateway* theGateway = nullptr;
Gateway::Gateway(void)
{
theGateway = this;
theMultiTaskProcess = this;
theProcess = this;
_packetEventQue.setMaxSize(MAX_INFLIGHTMESSAGES * MAX_CLIENTS);
_clientList = new ClientList(this);
_adapterManager = new AdapterManager(this);
_topics = new Topics();
@@ -95,15 +95,21 @@ Gateway::~Gateway()
{
free(_params.configName);
}
if (_params.qosMinusClientListName)
{
free(_params.qosMinusClientListName);
}
if (_params.bleAddress)
if (_params.rfcommAddr)
{
free(_params.bleAddress);
free(_params.rfcommAddr);
}
if (_params.gwCertskey)
{
free(_params.gwCertskey);
}
if (_params.gwPrivatekey)
{
free(_params.gwPrivatekey);
}
if (_adapterManager)
@@ -114,7 +120,6 @@ Gateway::~Gateway()
{
delete _clientList;
}
if (_topics)
{
delete _topics;
@@ -178,6 +183,14 @@ void Gateway::initialize(int argc, char** argv)
{
_params.rootCAfile = strdup(param);
}
if (getParam("DtlsCertsKey", param) == 0)
{
_params.gwCertskey = strdup(param);
}
if (getParam("DtlsPrivKey", param) == 0)
{
_params.gwPrivatekey = strdup(param);
}
if (getParam("GatewayID", param) == 0)
{
@@ -282,11 +295,14 @@ void Gateway::initialize(int argc, char** argv)
_params.maxClients = atoi(param);
}
if (getParam("BleAddress", param) == 0)
if (getParam("RFCOMMAddress", param) == 0)
{
_params.bleAddress = strdup(param);
_params.rfcommAddr = strdup(param);
}
/* Setup max PacketEventQue size */
_packetEventQue.setMaxSize(_params.maxInflightMsgs * _params.maxClients);
/* Initialize adapters */
_adapterManager->initialize(_params.gatewayName, _params.aggregatingGw, _params.forwarder, _params.qosMinus1);
@@ -308,26 +324,30 @@ void Gateway::run(void)
WRITELOG(" *\n%s\n", PAHO_COPYRIGHT3);
WRITELOG(" * Version: %s\n", PAHO_GATEWAY_VERSION);
WRITELOG("%s\n", PAHO_COPYRIGHT4);
WRITELOG("\n%s %s has been started.\n\n", currentDateTime(), _params.gatewayName);
WRITELOG(" ConfigFile: %s\n", _params.configName);
WRITELOG(" ConfigFile : %s\n", _params.configName);
if (_params.clientListName)
{
WRITELOG(" ClientList: %s\n", _params.clientListName);
WRITELOG(" ClientList : %s\n", _params.clientListName);
}
if (_params.predefinedTopicFileName)
{
WRITELOG(" PreDefFile: %s\n", _params.predefinedTopicFileName);
WRITELOG(" PreDefFile : %s\n", _params.predefinedTopicFileName);
}
WRITELOG(" SensorN/W: %s\n", _sensorNetwork.getDescription());
WRITELOG(" Broker: %s : %s, %s\n", _params.brokerName, _params.port, _params.portSecure);
WRITELOG(" Max number of Clients: %d\n", _params.maxClients);
WRITELOG(" RootCApath: %s\n", _params.rootCApath);
WRITELOG(" RootCAfile: %s\n", _params.rootCAfile);
WRITELOG(" CertKey: %s\n", _params.certKey);
WRITELOG(" PrivateKey: %s\n\n\n", _params.privateKey);
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(" CertKey : %s\n", _params.certKey);
WRITELOG(" PrivateKey : %s\n", _params.privateKey);
WRITELOG(" SensorN/W : %s\n", _sensorNetwork.getDescription());
#ifdef DTLS
WRITELOG(" DtlsCertsKey: %s\n", _params.gwCertskey);
WRITELOG(" DtlsPrivKey : %s\n", _params.gwPrivatekey);
#endif
WRITELOG(" Max Clients : %d\n\n", _params.maxClients);
WRITELOG("%s %s starts running.\n\n", currentDateTime(), _params.gatewayName);
_stopFlg = false;
@@ -408,12 +428,12 @@ bool Gateway::hasSecureConnection(void)
{
return (_params.certKey && _params.privateKey && _params.rootCApath && _params.rootCAfile);
}
/*=====================================
Class EventQue
=====================================*/
EventQue::EventQue()
{
}
EventQue::~EventQue()

7
MQTTSNGateway/src/MQTTSNGateway.h
View File

@@ -168,7 +168,9 @@ public:
bool qosMinus1 { false };
bool forwarder { false };
int maxClients {0};
char* bleAddress { nullptr };
char* rfcommAddr { nullptr };
char* gwCertskey { nullptr };
char* gwPrivatekey { nullptr };
};
/*=====================================
@@ -197,10 +199,10 @@ public:
int getParam(const char* parameter, char* value);
char* getClientListFileName(void);
char* getPredefinedTopicFileName(void);
bool hasSecureConnection(void);
Topics* getTopics(void);
bool IsStopping(void);
void requestSensorNetSubTask(void);
private:
GatewayParams _params;
@@ -214,7 +216,6 @@ private:
Topics* _topics;
bool _stopFlg;
};
}
#endif /* MQTTSNGATEWAY_H_ */

4
MQTTSNGateway/src/linux/Network.cpp
View File

@@ -656,3 +656,7 @@ bool Network::isSecure()
return _secureFlg;
}
void Network::setSecure(bool secureFlg)
{
_secureFlg = secureFlg;
}

1
MQTTSNGateway/src/linux/Network.h
View File

@@ -80,6 +80,7 @@ public:
bool isValid(void);
bool isSecure(void);
int getSock(void);
void setSecure(bool secureFlg);
private:
static SSL_CTX* _ctx;

1437
MQTTSNGateway/src/linux/dtls/SensorNetwork.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

163
MQTTSNGateway/src/linux/dtls/SensorNetwork.h Normal file
View File

@@ -0,0 +1,163 @@
/**************************************************************************************
* Copyright (c) 2021, Tomoaki Yamaguchi
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Eclipse Distribution License v1.0 which accompany this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
* and the Eclipse Distribution License is available at
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* Contributors:
* Tomoaki Yamaguchi - initial API and implementation and/or initial documentation
**************************************************************************************/
#ifndef SENSORNETWORK_H_
#define SENSORNETWORK_H_
#include "MQTTSNGWDefines.h"
#include "Threading.h"
#include <netinet/ip.h>
#include <openssl/ssl.h>
#include <openssl/err.h>
#include <string>
#include <poll.h>
using namespace std;
namespace MQTTSNGW
{
/*===========================================
Class SensorNetAddreess
============================================*/
typedef struct
{
int af;
union
{
struct in_addr ad4;
struct in6_addr ad6;
} addr;
} ipAddr_t;
class SensorNetAddress
{
public:
SensorNetAddress();
~SensorNetAddress();
void setAddress(ipAddr_t *Address, uint16_t port);
int setAddress(string *ipAddrPort);
int setIpAddress(string *IpAddress);
void setFamily(int type);
int getFamily(void);
void setPort(in_port_t port);
void setSockaddr4(sockaddr_in *sockaddr);
void setSockaddr6(sockaddr_in6 *sockaddr);
void cpyAddr4(sockaddr_in *sockaddr);
void cpyAddr6(sockaddr_in6 *sockaddr);
in_port_t getPort(void);
ipAddr_t* getIpAddress(void);
void setIndex(int index);
int getIndex(void);
void clear(void);
bool isMatch(SensorNetAddress *addr);
SensorNetAddress& operator =(SensorNetAddress &addr);
char* sprint(char *buf);
private:
int _pfdsIndex;
in_port_t _portNo;
ipAddr_t _ipAddr;
};
/*===========================================
Class Connections
============================================*/
#define POLL_UCAST 0
#define POLL_MCAST 1
#define POLL_SSL 2
typedef struct
{
int af;
SSL *ssl;
} afSSL_t;
class Connections
{
public:
Connections();
~Connections();
void initialize(int maxClient);
void close(int index);
int poll(int timeout);
int addClientSock(int sock);
int addClientSSL(SSL *ssl, int sock);
void setSockMulticast(int sock);
void setSockUnicast(int sock);
int getNumOfConnections(void);
int getNumOfClients(void);
SSL* getClientSSL(int index);
int getEventClient(int index);
int getSockClient(int index);
int getSockMulticast(void);
int getSockUnicast(void);
int getEventMulticast(void);
int getEventUnicast(void);
int getEventListen(void);
void closeSSL(int index);
int searchClient(SensorNetAddress *addr);
private:
pollfd *_pollfds;
SSL **_ssls;
SensorNetAddress **_clientAddr;
int _maxfds;
int _numfds;
Mutex _mutex;
};
/*===========================================
Class SensorNetwork
============================================*/
class SensorNetwork
{
friend class SensorNetSubTask;
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);
void initialize(void);
const char* getDescription(void);
SensorNetAddress* getSenderAddress(void);
void close();
private:
int openV4(string *ipAddress, uint16_t multiPortNo, uint16_t uniPortNo, uint16_t listenPortNo, uint32_t ttl);
int openV6(string *ipAddress, string *interface, uint16_t multiPortNo, uint16_t uniPortNo, uint16_t listenPortNo,
uint32_t hops);
int multicastRecv(uint8_t *buf, uint16_t len);
int getSendClient(int index, SensorNetAddress *addr);
int getSenderAddress(int sock, SensorNetAddress *addr);
int getUnicastClient(SensorNetAddress *addr);
void clearRecvData(int sock);
Mutex _mutex;
SensorNetAddress *_senderAddr;
SensorNetAddress *_multicastAddr;
SensorNetAddress *_unicastAddr;
string _description;
SSL_CTX *_dtlsctx;
Connections *_conns;
sockaddr_in _serverAddr4;
sockaddr_in6 _serverAddr6;
int _af;
};
}
#endif /* SENSORNETWORK_H_ */

18
MQTTSNGateway/src/linux/loralink/SensorNetwork.cpp
View File

@@ -383,7 +383,7 @@ bool LoRaLink::readApiFrame(LoRaLinkFrame_t* api, LoRaLinkReadParameters_t* para
int LoRaLink::send(LoRaLinkPayloadType_t type, const uint8_t* payload, uint16_t pLen, SensorNetAddress* addr)
{
D_NWSTACK("\r\n===> Send: ");
D_LRSTACK("\r\n===> Send: ");
uint8_t buf[2] = { 0 };
uint8_t chks = 0;
uint16_t len = pLen + 3; // 3 = DestAddr[1] + PayloadType[1] + Crc[1]
@@ -404,7 +404,7 @@ int LoRaLink::send(LoRaLinkPayloadType_t type, const uint8_t* payload, uint16_t
send(type);
chks += type;
D_NWSTACK("\r\n Payload: ");
D_LRSTACK("\r\n Payload: ");
for ( uint8_t i = 0; i < pLen; i++ ){
send(payload[i]); // Payload
@@ -412,21 +412,21 @@ int LoRaLink::send(LoRaLinkPayloadType_t type, const uint8_t* payload, uint16_t
}
chks = 0xff - chks;
D_NWSTACK(" checksum ");
D_LRSTACK(" checksum ");
send(chks);
D_NWSTACK("\r\n");
D_LRSTACK("\r\n");
/* wait ACK */
_sem.timedwait(LORALINK_TIMEOUT_ACK);
if ( _respCd == LORALINK_NO_FREE_CH )
{
D_NWSTACK(" Channel isn't free\r\n");
D_LRSTACK(" Channel isn't free\r\n");
return -1;
}
else if ( _respCd != LORALINK_ACK )
{
D_NWSTACK(" Not Acknowleged\r\n");
D_LRSTACK(" Not Acknowleged\r\n");
return -1;
}
return (int)pLen;
@@ -479,7 +479,7 @@ int LoRaLink::recv(uint8_t* buf)
/*
if ( *buf == ESCAPE )
{
D_NWSTACK( " %02x",buf[0] );
D_LRSTACK( " %02x",buf[0] );
if ( read(fd, buf, 1) == 1 )
{
*buf = PAD ^ *buf;
@@ -491,7 +491,7 @@ int LoRaLink::recv(uint8_t* buf)
}
*/
D_NWSTACK( " %02x",buf[0] );
D_LRSTACK(" %02x", buf[0]);
return 0;
}
}
@@ -552,7 +552,7 @@ bool SerialPort::send(unsigned char b)
}
else
{
D_NWSTACK( " %02x", b);
D_LRSTACK(" %02x", b);
return true;
}
}

9
MQTTSNGateway/src/linux/loralink/SensorNetwork.h
View File

@@ -11,7 +11,7 @@
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* Contributors:
* Tomoaki Yamaguchi - initial API and implementation
* Tomoaki Yamaguchi - initial API and implementation
**************************************************************************************/
#ifndef SENSORNETWORKX_H_
#define SENSORNETWORKX_H_
@@ -25,12 +25,11 @@ using namespace std;
namespace MQTTSNGW
{
//#define DEBUG_NWSTACK
#ifdef DEBUG_NWSTACK
#define D_NWSTACK(...) printf(__VA_ARGS__); fflush(stdout)
#ifdef DEBUG_NW
#define D_LRSTACK(...) printf(__VA_ARGS__); fflush(stdout)
#else
#define D_NWSTACK(...)
#define D_LRSTACK(...)
#endif

6
MQTTSNGateway/src/linux/rfcomm/SensorNetwork.h
View File

@@ -26,12 +26,6 @@ using namespace std;
namespace MQTTSNGW
{
#ifdef DEBUG_NWSTACK
#define D_NWSTACK(...) printf(__VA_ARGS__)
#else
#define D_NWSTACK(...)
#endif
#define MAX_RFCOMM_CH 30
/*===========================================

311
MQTTSNGateway/src/linux/udp/SensorNetwork.cpp
View File

@@ -20,10 +20,13 @@
#include <netinet/ip.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <net/if.h>
#include <sys/ioctl.h>
#include <string.h>
#include <regex>
#include <string>
#include <stdlib.h>
#include <poll.h>
#include "SensorNetwork.h"
#include "MQTTSNGWProcess.h"
@@ -196,33 +199,33 @@ void SensorNetwork::initialize(void)
* MulticastIP=225.1.1.1
* MulticastPortNo=1883
*
*/
if (theProcess->getParam("MulticastIP", param) == 0)
{
ip = param;
_description = "UDP Multicast ";
_description += param;
}
if (theProcess->getParam("MulticastPortNo", param) == 0)
{
multicastPortNo = atoi(param);
_description += ":";
_description += param;
}
if (theProcess->getParam("GatewayPortNo", param) == 0)
{
unicastPortNo = atoi(param);
_description += " Gateway Port ";
_description += param;
}
if (theProcess->getParam("MulticastTTL", param) == 0)
{
ttl = atoi(param);
_description += " TTL: ";
_description += param;
}
*/
if (theProcess->getParam("MulticastIP", param) == 0)
{
ip = param;
_description = "UDP Multicast ";
_description += param;
}
if (theProcess->getParam("MulticastPortNo", param) == 0)
{
multicastPortNo = atoi(param);
_description += ":";
_description += param;
}
if (theProcess->getParam("GatewayPortNo", param) == 0)
{
unicastPortNo = atoi(param);
_description += ", Gateway Port:";
_description += param;
}
if (theProcess->getParam("MulticastTTL", param) == 0)
{
ttl = atoi(param);
_description += ", TTL:";
_description += param;
}
/* Prepare UDP sockets */
/* setup UDP sockets */
errno = 0;
if ( UDPPort::open(ip.c_str(), multicastPortNo, unicastPortNo, ttl) < 0 )
{
@@ -247,9 +250,7 @@ SensorNetAddress* SensorNetwork::getSenderAddress(void)
UDPPort::UDPPort()
{
_disconReq = false;
_sockfdUnicast = -1;
_sockfdMulticast = -1;
_ttl = 0;
memset(_pollFds, 0, sizeof(_pollFds));
}
UDPPort::~UDPPort()
@@ -259,130 +260,126 @@ UDPPort::~UDPPort()
void UDPPort::close(void)
{
if (_sockfdUnicast > 0)
{
::close(_sockfdUnicast);
_sockfdUnicast = -1;
}
if (_sockfdMulticast > 0)
{
::close(_sockfdMulticast);
_sockfdMulticast = -1;
}
for (int i = 0; i < 2; i++)
{
if (_pollFds[i].fd > 0)
{
::close(_pollFds[i].fd);
_pollFds[i].fd = 0;
}
}
}
int UDPPort::open(const char* ipAddress, uint16_t multiPortNo, uint16_t uniPortNo, unsigned int ttl)
int UDPPort::open(const char *multicastIP, uint16_t multiPortNo, uint16_t uniPortNo, unsigned int ttl)
{
char loopch = 0;
const int reuse = 1;
int optval = 0;
int sock = 0;
if (uniPortNo == 0 || multiPortNo == 0)
{
D_NWSTACK("error portNo undefined in UDPPort::open\n");
return -1;
}
if (uniPortNo == 0 || multiPortNo == 0)
{
D_NWSTACK("error portNo undefined in UDPPort::open\n");
return -1;
}
uint32_t ip = inet_addr(ipAddress);
_multicastAddr.setAddress(ip, htons(multiPortNo));
_unicastAddr.setAddress(ip, htons(uniPortNo));
_ttl = ttl;
/*------ Create unicast socket --------*/
sock = socket(AF_INET, SOCK_DGRAM, 0);
if (sock < 0)
{
D_NWSTACK("error can't create unicast socket in UDPPort::open\n");
return -1;
}
/*------ Create unicast socket --------*/
_sockfdUnicast = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
if (_sockfdUnicast < 0)
{
D_NWSTACK("error can't create unicast socket in UDPPort::open\n");
return -1;
}
sockaddr_in addru;
addru.sin_family = AF_INET;
addru.sin_port = htons(uniPortNo);
addru.sin_addr.s_addr = INADDR_ANY;
setsockopt(_sockfdUnicast, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
sockaddr_in addru;
addru.sin_family = AF_INET;
addru.sin_port = htons(uniPortNo);
addru.sin_addr.s_addr = INADDR_ANY;
if (::bind(_sockfdUnicast, (sockaddr*) &addru, sizeof(addru)) < 0)
if (::bind(sock, (sockaddr*) &addru, sizeof(addru)) < 0)
{
D_NWSTACK("error can't bind unicast socket in UDPPort::open\n");
return -1;
}
if (setsockopt(_sockfdUnicast, IPPROTO_IP, IP_MULTICAST_LOOP, (char*) &loopch, sizeof(loopch)) < 0)
{
D_NWSTACK("error IP_MULTICAST_LOOP in UDPPort::open\n");
close();
return -1;
}
/*------ Create Multicast socket --------*/
_sockfdMulticast = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
if (_sockfdMulticast < 0)
_pollFds[0].fd = sock;
_pollFds[0].events = POLLIN;
/*------ Create Multicast socket --------*/
sock = socket(AF_INET, SOCK_DGRAM, 0);
if (sock < 0)
{
D_NWSTACK("error can't create multicast socket in UDPPort::open\n");
close();
return -1;
}
setsockopt(_sockfdMulticast, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
optval = 1;
setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));
sockaddr_in addrm;
addrm.sin_family = AF_INET;
addrm.sin_port = _multicastAddr.getPortNo();
addrm.sin_addr.s_addr = INADDR_ANY;
sockaddr_in addrm;
addrm.sin_family = AF_INET;
addrm.sin_port = htons(multiPortNo);
addrm.sin_addr.s_addr = INADDR_ANY;
if (::bind(_sockfdMulticast, (sockaddr*) &addrm, sizeof(addrm)) < 0)
{
D_NWSTACK("error can't bind multicast socket in UDPPort::open\n");
return -1;
}
if (setsockopt(_sockfdMulticast, IPPROTO_IP, IP_MULTICAST_LOOP, (char*) &loopch, sizeof(loopch)) < 0)
{
D_NWSTACK("error IP_MULTICAST_LOOP in UDPPort::open\n");
close();
return -1;
}
if (::bind(sock, (sockaddr*) &addrm, sizeof(addrm)) < 0)
{
D_NWSTACK("error can't bind multicast socket in UDPPort::open\n");
return -1;
}
ip_mreq mreq;
mreq.imr_interface.s_addr = INADDR_ANY;
mreq.imr_multiaddr.s_addr = _multicastAddr.getIpAddress();
ip_mreq mreq;
memset(&mreq, 0, sizeof(mreq));
mreq.imr_interface.s_addr = INADDR_ANY;
mreq.imr_multiaddr.s_addr = inet_addr(multicastIP);
if (setsockopt(_sockfdMulticast, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq)) < 0)
{
D_NWSTACK("error Multicast IP_ADD_MEMBERSHIP in UDPPort::open\n");
close();
return -1;
}
if (setsockopt(sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq)) < 0)
{
D_NWSTACK("error Multicast IP_ADD_MEMBERSHIP in UDPPort::open\n");
close();
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(sock, IPPROTO_IP, IP_MULTICAST_TTL, &ttl, sizeof(ttl)) < 0)
{
D_NWSTACK("error Multicast IP_MULTICAST_TTL 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");
close();
return -1;
}
return 0;
#ifdef DEBUG_NW
optval = 1;
#else
optval = 0;
#endif
if (setsockopt(sock, IPPROTO_IP, IP_MULTICAST_LOOP, &optval, sizeof(optval)) < 0)
{
D_NWSTACK("error IP_MULTICAST_LOOP in UDPPort::open\n");
close();
return -1;
}
_multicastAddr.setAddress(inet_addr(multicastIP), htons(multiPortNo));
_pollFds[1].fd = sock;
_pollFds[1].events = POLLIN;
return 0;
}
int UDPPort::unicast(const uint8_t* buf, uint32_t length, SensorNetAddress* addr)
{
sockaddr_in dest;
dest.sin_family = AF_INET;
dest.sin_port = addr->getPortNo();
dest.sin_addr.s_addr = addr->getIpAddress();
sockaddr_in dest;
dest.sin_family = AF_INET;
dest.sin_port = addr->getPortNo();
dest.sin_addr.s_addr = addr->getIpAddress();
int status = ::sendto(_sockfdUnicast, buf, length, 0, (const sockaddr*) &dest, sizeof(dest));
if (status < 0)
{
D_NWSTACK("errno == %d in UDPPort::sendto\n", errno);
}
D_NWSTACK("sendto %s:%u length = %d\n", inet_ntoa(dest.sin_addr), ntohs(dest.sin_port), status);
return status;
int status = ::sendto(_pollFds[0].fd, buf, length, 0, (const sockaddr*) &dest, sizeof(dest));
if (status < 0)
{
D_NWSTACK("errno == %d in UDPPort::sendto\n", errno);
}
D_NWSTACK("sendto %s:%u length = %d\n", inet_ntoa(dest.sin_addr), ntohs(dest.sin_port), status);
return status;
}
int UDPPort::broadcast(const uint8_t* buf, uint32_t length)
@@ -392,55 +389,35 @@ int UDPPort::broadcast(const uint8_t* buf, uint32_t length)
int UDPPort::recv(uint8_t* buf, uint16_t len, SensorNetAddress* addr)
{
struct timeval timeout;
fd_set recvfds;
int maxSock = 0;
int rc = 0;
poll(_pollFds, 2, 2000); // Timeout 2 seconds
timeout.tv_sec = 1;
timeout.tv_usec = 0; // 1 sec
FD_ZERO(&recvfds);
FD_SET(_sockfdUnicast, &recvfds);
FD_SET(_sockfdMulticast, &recvfds);
if (_sockfdMulticast > _sockfdUnicast)
{
maxSock = _sockfdMulticast;
}
else
{
maxSock = _sockfdUnicast;
}
int rc = 0;
if ( select(maxSock + 1, &recvfds, 0, 0, &timeout) > 0 )
{
if (FD_ISSET(_sockfdUnicast, &recvfds))
{
rc = recvfrom(_sockfdUnicast, buf, len, 0, addr);
}
else if (FD_ISSET(_sockfdMulticast, &recvfds))
{
rc = recvfrom(_sockfdMulticast, buf, len, 0, &_multicastAddr);
}
}
return rc;
if (_pollFds[0].revents == POLLIN)
{
rc = recvfrom(_pollFds[0].fd, buf, len, 0, addr);
}
else if (_pollFds[1].revents == POLLIN)
{
rc = recvfrom(_pollFds[1].fd, buf, len, 0, addr);
}
return rc;
}
int UDPPort::recvfrom(int sockfd, uint8_t* buf, uint16_t len, uint8_t flags, SensorNetAddress* addr)
{
sockaddr_in sender;
socklen_t addrlen = sizeof(sender);
memset(&sender, 0, addrlen);
sockaddr_in sender;
socklen_t addrlen = sizeof(sender);
memset(&sender, 0, addrlen);
int status = ::recvfrom(sockfd, buf, len, flags, (sockaddr*) &sender, &addrlen);
int status = ::recvfrom(sockfd, buf, len, flags, (sockaddr*) &sender, &addrlen);
if (status < 0 && errno != EAGAIN)
{
D_NWSTACK("errno == %d in UDPPort::recvfrom\n", errno);
return -1;
}
addr->setAddress(sender.sin_addr.s_addr, sender.sin_port);
D_NWSTACK("recved from %s:%d length = %d\n", inet_ntoa(sender.sin_addr),ntohs(sender.sin_port), status);
return status;
if (status < 0 && errno != EAGAIN)
{
D_NWSTACK("errno == %d in UDPPort::recvfrom\n", errno);
return -1;
}
addr->setAddress(sender.sin_addr.s_addr, sender.sin_port);
D_NWSTACK("recved from %s:%d length = %d\n", inet_ntoa(sender.sin_addr),ntohs(sender.sin_port), status);
return status;
}

15
MQTTSNGateway/src/linux/udp/SensorNetwork.h
View File

@@ -19,18 +19,13 @@
#include "MQTTSNGWDefines.h"
#include <string>
#include <poll.h>
using namespace std;
namespace MQTTSNGW
{
#ifdef DEBUG_NWSTACK
#define D_NWSTACK(...) printf(__VA_ARGS__)
#else
#define D_NWSTACK(...)
#endif
/*===========================================
Class SensorNetAddreess
============================================*/
@@ -70,13 +65,9 @@ private:
void setNonBlocking(const bool);
int recvfrom(int sockfd, uint8_t* buf, uint16_t len, uint8_t flags, SensorNetAddress* addr);
int _sockfdUnicast;
int _sockfdMulticast;
SensorNetAddress _multicastAddr;
SensorNetAddress _unicastAddr;
pollfd _pollFds[2];
bool _disconReq;
unsigned int _ttl;
SensorNetAddress _multicastAddr;
};
/*===========================================

607
MQTTSNGateway/src/linux/udp6/SensorNetwork.cpp
View File

@@ -1,6 +1,6 @@
/**************************************************************************************
* Copyright (c) 2017, Benjamin Aigner
* Copyright (c) 2016, Tomoaki Yamaguchi (original UDPv4 implementation)
* Copyright (c) 2021, Tomoaki Yamaguchi
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
@@ -20,8 +20,10 @@
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/ioctl.h>
#include <netinet/ip.h>
#include <net/if.h>
#include <ifaddrs.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <string.h>
@@ -40,61 +42,57 @@ using namespace MQTTSNGW;
============================================*/
SensorNetAddress::SensorNetAddress()
{
_portNo = 0;
memset((void *)&_IpAddr,0,sizeof(_IpAddr));
memset((void*) &_IpAddr, 0, sizeof(_IpAddr));
}
SensorNetAddress::~SensorNetAddress()
{
}
struct sockaddr_in6 *SensorNetAddress::getIpAddress(void)
sockaddr_in6* SensorNetAddress::getIpAddress(void)
{
return &_IpAddr;
return &_IpAddr;
}
uint16_t SensorNetAddress::getPortNo(void)
{
return _portNo;
return _IpAddr.sin6_port;
}
void SensorNetAddress::setAddress(struct sockaddr_in6 *IpAddr, uint16_t port)
void SensorNetAddress::setAddress(struct sockaddr_in6 *IpAddr)
{
memcpy((void *)&_IpAddr,IpAddr,sizeof(_IpAddr));
_portNo = port;
memcpy((void*) &_IpAddr, IpAddr, sizeof(_IpAddr));
}
/**
* convert Text data to SensorNetAddress
* @param data is a IPV6_Address:PortNo format string
* @param data is a string [IPV6_Address]:PortNo
* @return success = 0, Invalid format = -1
*/
int SensorNetAddress::setAddress(string* data)
{
size_t pos = data->find_last_of("]:");
size_t pos = data->find_last_of(":");
if (pos != string::npos)
{
int portNo = 0;
string port = data->substr(pos + 1);
if ( pos != string::npos)
{
int portNo = 0;
string port = data->substr(pos + 1);
if ((portNo = atoi(port.c_str())) > 0)
{
_IpAddr.sin6_port = htons(portNo);
_IpAddr.sin6_family = AF_INET6;
string ip = data->substr(1, pos - 2);
const char *cstr = ip.c_str();
if ( ( portNo = atoi(port.c_str()) ) > 0 )
{
_portNo = htons(portNo);
string ip = data->substr(1,pos - 1);
const char *cstr = ip.c_str();
if (inet_pton(AF_INET6, cstr, &(_IpAddr.sin6_addr)) == 1 )
{
return 0;
}
}
}
_portNo = 0;
memset((void *)&_IpAddr,0,sizeof(_IpAddr));
return -1;
if (inet_pton(AF_INET6, cstr, &(_IpAddr.sin6_addr)) == 1)
{
return 0;
}
}
}
memset((void*) &_IpAddr, 0, sizeof(_IpAddr));
return -1;
}
/**
@@ -104,43 +102,42 @@ int SensorNetAddress::setAddress(string* data)
*/
int SensorNetAddress::setAddress(const char* data)
{
if ( inet_pton(AF_INET6, data, &(_IpAddr.sin6_addr)) == 1 )
{
return 0;
}
else
{
return -1;
}
if (inet_pton(AF_INET6, data, &(_IpAddr.sin6_addr)) == 1)
{
_IpAddr.sin6_family = AF_INET6;
return 0;
}
else
{
return -1;
}
}
char* SensorNetAddress::getAddress(void)
{
inet_ntop(AF_INET6, &(_IpAddr.sin6_addr), _addrString, INET6_ADDRSTRLEN);
return _addrString;
inet_ntop(AF_INET6, &(_IpAddr.sin6_addr), _addrString, INET6_ADDRSTRLEN);
return _addrString;
}
bool SensorNetAddress::isMatch(SensorNetAddress* addr)
{
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);
return (this->_IpAddr.sin6_port == addr->_IpAddr.sin6_port)
&& (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)
{
this->_portNo = addr._portNo;
memcpy(&this->_IpAddr.sin6_addr, &addr._IpAddr.sin6_addr, sizeof(this->_IpAddr.sin6_addr));
return *this;
memcpy(&this->_IpAddr, &addr._IpAddr, sizeof(this->_IpAddr));
return *this;
}
char* SensorNetAddress::sprint(char* buf)
{
char ip[INET6_ADDRSTRLEN];
inet_ntop(AF_INET6, &(_IpAddr.sin6_addr), ip, INET6_ADDRSTRLEN);
sprintf( buf, "%s:", ip);
sprintf( buf + strlen(buf), "%d", ntohs(_portNo));
return buf;
sprintf(buf, "[%s]:", getAddress());
sprintf(buf + strlen(buf), "%d", ntohs(_IpAddr.sin6_port));
return buf;
}
/*===========================================
@@ -156,75 +153,74 @@ SensorNetwork::~SensorNetwork()
int SensorNetwork::unicast(const uint8_t* payload, uint16_t payloadLength, SensorNetAddress* sendToAddr)
{
return UDPPort6::unicast(payload, payloadLength, sendToAddr);
return UDPPort6::unicast(payload, payloadLength, sendToAddr);
}
int SensorNetwork::broadcast(const uint8_t* payload, uint16_t payloadLength)
{
return UDPPort6::broadcast(payload, payloadLength);
return UDPPort6::broadcast(payload, payloadLength);
}
int SensorNetwork::read(uint8_t* buf, uint16_t bufLen)
{
return UDPPort6::recv(buf, bufLen, &_clientAddr);
return UDPPort6::recv(buf, bufLen, &_clientAddr);
}
void SensorNetwork::initialize(void)
{
char param[MQTTSNGW_PARAM_MAX];
uint16_t unicastPortNo = 0;
string ip;
string broadcast;
string interface;
unsigned int hops = 1;
char param[MQTTSNGW_PARAM_MAX];
uint16_t unicastPortNo = 0;
uint16_t multicastPortNo = 0;
string ip;
string multicast;
string interface;
uint32_t hops = 1;
if (theProcess->getParam("GatewayUDP6Bind", param) == 0)
{
ip = param;
_description = "GatewayUDP6Bind: ";
_description += param;
}
if (theProcess->getParam("GatewayUDP6Port", param) == 0)
{
unicastPortNo = atoi(param);
_description += " Gateway Port: ";
_description += param;
}
if (theProcess->getParam("GatewayUDP6Broadcast", param) == 0)
{
broadcast = param;
_description += " Broadcast Address: ";
_description += param;
}
if (theProcess->getParam("GatewayUDP6If", param) == 0)
{
interface = param;
_description += " Interface: ";
_description += param;
}
if (theProcess->getParam("GatewayUDP6Hops", param) == 0)
{
hops = atoi(param);
_description += " Hops: ";
_description += param;
}
if (theProcess->getParam("MulticastIPv6", param) == 0)
{
multicast = param;
_description += "Multicast Address: [";
_description += param;
}
if (theProcess->getParam("MulticastIPv6PortNo", param) == 0)
{
multicastPortNo = atoi(param);
_description += "]:";
_description += param;
}
if (theProcess->getParam("GatewayIPv6PortNo", param) == 0)
{
unicastPortNo = atoi(param);
_description += ", Gateway Port:";
_description += param;
}
if (theProcess->getParam("MulticastIPv6If", param) == 0)
{
interface = param;
_description += ", Interface: ";
_description += param;
}
if (theProcess->getParam("MulticastHops", param) == 0)
{
hops = atoi(param);
_description += ", Hops:";
_description += param;
}
errno = 0;
if ( UDPPort6::open(ip.c_str(), unicastPortNo, broadcast.c_str(), interface.c_str(), hops) < 0 )
{
throw EXCEPTION("Can't open a UDP6", errno);
}
if (UDPPort6::open(unicastPortNo, multicastPortNo, multicast.c_str(), interface.c_str(), hops) < 0)
{
throw EXCEPTION("Can't open a UDP6", errno);
}
}
const char* SensorNetwork::getDescription(void)
{
return _description.c_str();
return _description.c_str();
}
SensorNetAddress* SensorNetwork::getSenderAddress(void)
{
return &_clientAddr;
return &_clientAddr;
}
/*=========================================
@@ -233,278 +229,241 @@ SensorNetAddress* SensorNetwork::getSenderAddress(void)
UDPPort6::UDPPort6()
{
_disconReq = false;
_sockfdUnicast = -1;
_sockfdMulticast = -1;
_disconReq = false;
_hops = 0;
}
UDPPort6::~UDPPort6()
{
close();
close();
}
void UDPPort6::close(void)
{
if (_sockfdUnicast > 0)
{
::close(_sockfdUnicast);
_sockfdUnicast = -1;
}
if (_sockfdMulticast > 0)
{
::close(_sockfdMulticast);
_sockfdMulticast = -1;
}
for (int i = 0; i < 2; i++)
{
if (_pollfds[i].fd > 0)
{
::close(_pollfds[i].fd);
_pollfds[i].fd = 0;
}
}
}
int UDPPort6::open(const char* ipAddress, uint16_t uniPortNo, const char* broadcastAddr, const char* interfaceName, unsigned int hops)
int UDPPort6::open(uint16_t uniPortNo, uint16_t multiPortNo, const char *multicastAddr, const char *interfaceName,
uint32_t hops)
{
struct addrinfo hints, *res;
int errnu;
const int reuse = 1;
int optval = 0;
int sock = 0;
sockaddr_in6 addr6;
uint32_t ifindex = 0;
if (uniPortNo == 0)
{
WRITELOG("error portNo undefined in UDPPort::open\n");
return -1;
}
errno = 0;
if (uniPortNo == 0 || multiPortNo == 0)
{
D_NWSTACK("error portNo undefined in UDPPort6::open\n");
return -1;
}
memset(&hints, 0, sizeof hints);
hints.ai_family = AF_INET6; // use IPv6
hints.ai_socktype = SOCK_DGRAM;
hints.ai_flags = AI_PASSIVE; //use local IF address
// Create a unicast socket
sock = socket(AF_INET6, SOCK_DGRAM, 0);
if (sock < 0)
{
D_NWSTACK("UDP6::open - unicast socket: %s", strerror(errno));
return -1;
}
getaddrinfo(NULL, std::to_string(uniPortNo).c_str(), &hints, &res);
_pollfds[0].fd = sock;
_pollfds[0].events = POLLIN;
_sockfdMulticast = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
if(_sockfdMulticast <0)
{
WRITELOG("UDP6::open - multicast: %s",strerror(_sockfdMulticast));
return errno;
}
optval = 1;
setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char*) &optval, sizeof(optval));
//select the interface
unsigned int ifindex;
ifindex = if_nametoindex(interfaceName);
errnu = setsockopt(_sockfdMulticast, IPPROTO_IPV6, IPV6_MULTICAST_IF, &ifindex,sizeof(ifindex));
if(errnu <0)
{
WRITELOG("UDP6::open - limit IF: %s",strerror(errnu));
return errnu;
}
optval = 1;
if (setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, (char*) &optval, sizeof(optval)) < 0)
{
D_NWSTACK("\033[0m\033[0;31m unicast socket error %s IPV6_V6ONLY\033[0m\033[0;37m\n", strerror(errno));
close();
return -1;
}
strcpy(_interfaceName,interfaceName);
if (setsockopt(sock, IPPROTO_IPV6, IPV6_UNICAST_HOPS, &hops, sizeof(hops)) < 0)
{
D_NWSTACK("\033[0m\033[0;31m error %s IPV6_UNICAST_HOPS\033[0m\033[0;37m\n", strerror(errno));
close();
return -1;
}
//restrict the socket to IPv6 only
int on = 1;
errnu = setsockopt(_sockfdMulticast, IPPROTO_IPV6, IPV6_V6ONLY, (char *)&on, sizeof(on));
if(errnu <0)
{
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
//according to: https://beej.us/guide/bgnet/output/html/multipage/
memset(&hints, 0, sizeof hints);
hints.ai_family = AF_INET6; // use IPv6, whichever
hints.ai_socktype = SOCK_DGRAM;
hints.ai_flags = AI_PASSIVE; // fill in my IP for me
//no specific address, bind to available ones...
getaddrinfo(NULL, std::to_string(uniPortNo).c_str(), &hints, &res);
//create the socket
_sockfdUnicast = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
if (_sockfdUnicast < 0)
{
WRITELOG("UDP6::open - unicast socket: %s",strerror(_sockfdUnicast));
return -1;
}
//if given, set a given device name to bind to
if(strlen(interfaceName) > 0)
{
if (strlen(interfaceName) > 0)
{
ifindex = if_nametoindex(interfaceName);
#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));
setsockopt(sock, IPPROTO_IP, IP_BOUND_IF, &ifindex, sizeof(ifindex));
#else
setsockopt(sock, SOL_SOCKET, SO_BINDTODEVICE, interfaceName, strlen(interfaceName));
#endif
}
}
//socket option: reuse address
setsockopt(_sockfdUnicast, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
memset(&addr6, 0, sizeof(addr6));
addr6.sin6_family = AF_INET6;
addr6.sin6_port = htons(uniPortNo);
addr6.sin6_addr = in6addr_any;
//finally: bind...
errnu = ::bind(_sockfdUnicast, res->ai_addr, res->ai_addrlen);
if (errnu < 0)
{
WRITELOG("error can't bind unicast socket in UDPPort::open: %s\n",strerror(errnu));
return -1;
}
if (::bind(sock, (sockaddr*) &addr6, sizeof(addr6)) < 0)
{
D_NWSTACK("error can't bind unicast socket in UDPPort6::open: %s\n", strerror(errno));
close();
return -1;
}
//if given, set a broadcast address; otherwise it will be ::
if(strlen(broadcastAddr) > 0)
{
_grpAddr.setAddress(broadcastAddr);
} else {
_grpAddr.setAddress("::");
}
//everything went fine...
freeaddrinfo(res);
return 0;
// create a MULTICAST socket
sock = socket(AF_INET6, SOCK_DGRAM, 0);
if (sock < 0)
{
D_NWSTACK("UDP6::open - multicast: %s", strerror(errno));
close();
return -1;
}
_pollfds[1].fd = sock;
_pollfds[1].events = POLLIN;
optval = 1;
if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (const char*) &optval, sizeof(optval)) < 0)
{
D_NWSTACK("\033[0m\033[0;31m multicast socket error %s SO_REUSEADDR\033[0m\033[0;37m\n", strerror(errno));
close();
return -1;
}
optval = 1;
if (setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, (char*) &optval, sizeof(optval)) < 0)
{
D_NWSTACK("\033[0m\033[0;31m multicast socket error %s IPV6_V6ONLY\033[0m\033[0;37m\n", strerror(errno));
close();
return -1;
}
memset(&addr6, 0, sizeof(addr6));
addr6.sin6_family = AF_INET6;
addr6.sin6_port = htons(multiPortNo);
addr6.sin6_addr = in6addr_any;
if (::bind(sock, (sockaddr*) &addr6, sizeof(addr6)) < 0)
{
close();
D_NWSTACK("error can't bind multicast socket in UDPPort6::open: %s\n", strerror(errno));
return -1;
}
ipv6_mreq addrm;
addrm.ipv6mr_interface = ifindex;
inet_pton(AF_INET6, multicastAddr, &addrm.ipv6mr_multiaddr);
if (setsockopt(sock, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, &addrm, sizeof(addrm)) < 0)
{
D_NWSTACK("\033[0m\033[0;31m error %d IPV6_ADD_MEMBERSHIP in Udp6Port::open\033[0m\033[0;37m\n", errno);
close();
return false;
}
#ifdef DEBUG_NW
optval = 1;
#else
optval = 0;
#endif
if (setsockopt(sock, IPPROTO_IPV6, IPV6_MULTICAST_LOOP, (char*) &optval, sizeof(optval)) < 0)
{
D_NWSTACK("\033[0m\033[0;31m error %s IPV6_MULTICAST_LOOP\033[0m\033[0;37m\n", strerror(errno));
close();
return false;
}
if (setsockopt(sock, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, &hops, sizeof(hops)) < 0)
{
D_NWSTACK("\033[0m\033[0;31m error %s IPV6_MULTICAST_HOPS\033[0m\033[0;37m\n", strerror(errno));
close();
return -1;
}
memcpy(&addr6.sin6_addr, &addrm.ipv6mr_multiaddr, sizeof(addrm.ipv6mr_multiaddr));
_grpAddr.setAddress(&addr6);
return 0;
}
int UDPPort6::unicast(const uint8_t* buf, uint32_t length, SensorNetAddress* addr)
{
char destStr[INET6_ADDRSTRLEN+10];
struct addrinfo hints, *res;
memset(&hints, 0, sizeof hints);
hints.ai_family = AF_INET6; // use IPv6
hints.ai_socktype = SOCK_DGRAM;
sockaddr_in6 dest;
memset(&dest, 0, sizeof(dest));
dest.sin6_family = AF_INET6;
dest.sin6_port = addr->getPortNo();
memcpy(dest.sin6_addr.s6_addr, (const void*) &addr->getIpAddress()->sin6_addr, sizeof(in6_addr));
int err = 0;
int port = 0;
string portStr;
if(addr->getPortNo() != 0)
{
port = htons(addr->getPortNo());
portStr = to_string(port);
} else {
port = _uniPortNo;
portStr = to_string(port);
}
#ifdef DEBUG_NW
char addrBuf[INET6_ADDRSTRLEN];
addr->sprint(addrBuf);
D_NWSTACK("sendto %s\n", addrBuf);
#endif
errno = 0;
int status = ::sendto(_pollfds[0].fd, buf, length, 0, (const sockaddr*) &dest, sizeof(dest));
if(strlen(_interfaceName) != 0)
{
strcpy(destStr, addr->getAddress());
strcat(destStr,"%");
strcat(destStr,_interfaceName);
if(IN6_IS_ADDR_LINKLOCAL(&addr->getIpAddress()->sin6_addr))
{
err = getaddrinfo(destStr, portStr.c_str(), &hints, &res);
}
else
{
err = getaddrinfo(addr->getAddress(), portStr.c_str(), &hints, &res);
}
} else {
strcpy(destStr, addr->getAddress());
err = getaddrinfo(addr->getAddress(), portStr.c_str(), &hints, &res);
}
if ( err != 0)
{
WRITELOG("UDP6::broadcast - getaddrinfo: %s",strerror(errno));
return err;
}
int status = ::sendto(_sockfdUnicast, buf, length, 0, res->ai_addr, res->ai_addrlen);
if (status < 0)
{
WRITELOG("errno in UDPPort::unicast(sendto): %d, %s\n",status,strerror(errno));
}
return status;
if (status < 0)
{
D_NWSTACK("%s in UDPPor6t::sendto\n", strerror(errno));
}
return status;
}
int UDPPort6::broadcast(const uint8_t* buf, uint32_t length)
{
struct addrinfo hint,*info;
int err;
memset( &hint, 0, sizeof( hint ) );
int err = unicast(buf, length, &_grpAddr);
hint.ai_family = AF_INET6;
hint.ai_socktype = SOCK_DGRAM;
hint.ai_protocol = 0;
if (err < 0)
{
D_NWSTACK("UDP6::broadcast - sendto: %s", strerror(errno));
return err;
}
errno = 0;
if(strlen(_interfaceName) != 0)
{
char destStr[80];
strcpy(destStr, _grpAddr.getAddress());
strcat(destStr,"%");
strcat(destStr,_interfaceName);
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 );
}
else
{
err = getaddrinfo(_grpAddr.getAddress(), std::to_string(_uniPortNo).c_str(), &hint, &info );
}
} else {
err = getaddrinfo(_grpAddr.getAddress(), std::to_string(_uniPortNo).c_str(), &hint, &info );
}
if( err != 0 ) {
WRITELOG("UDP6::broadcast - getaddrinfo: %s",strerror(errno));
return err;
}
err = sendto(_sockfdMulticast, buf, length, 0, info->ai_addr, info->ai_addrlen );
if(err < 0 ) {
WRITELOG("UDP6::broadcast - sendto: %s",strerror(errno));
return err;
}
return 0;
return 0;
}
int UDPPort6::recv(uint8_t* buf, uint16_t len, SensorNetAddress* addr)
{
struct timeval timeout;
fd_set recvfds;
int rc = poll(_pollfds, 2, 2000); // Timeout 2secs
if (rc == 0)
{
return rc;
}
timeout.tv_sec = 1;
timeout.tv_usec = 0; // 1 sec
FD_ZERO(&recvfds);
FD_SET(_sockfdUnicast, &recvfds);
int rc = 0;
if ( select(_sockfdUnicast + 1, &recvfds, 0, 0, &timeout) > 0 )
{
if (FD_ISSET(_sockfdUnicast, &recvfds))
{
rc = recvfrom(_sockfdUnicast, buf, len, 0, addr);
}
}
return rc;
for (int i = 0; i < 2; i++)
{
if (_pollfds[i].revents & POLLIN)
{
return recvfrom(_pollfds[i].fd, buf, len, 0, addr);
}
}
return 0;
}
int UDPPort6::recvfrom(int sockfd, uint8_t* buf, uint16_t len, uint8_t flags, SensorNetAddress* addr)
{
sockaddr_in6 sender;
socklen_t addrlen = sizeof(sender);
memset(&sender, 0, addrlen);
sockaddr_in6 sender;
socklen_t addrlen = sizeof(sender);
memset(&sender, 0, addrlen);
int status = ::recvfrom(sockfd, buf, len, flags, (sockaddr*) &sender, &addrlen);
int status = ::recvfrom(sockfd, buf, len, flags, (sockaddr*) &sender, &addrlen);
if (status < 0 && errno != EAGAIN)
{
WRITELOG("errno == %d in UDPPort::recvfrom: %s\n",errno,strerror(errno));
return -1;
}
addr->setAddress(&sender, (uint16_t)sender.sin6_port);
//D_NWSTACK("recved from %s:%d length = %d\n", inet_ntoa(sender.sin_addr),ntohs(sender.sin_port), status);
return status;
if (status < 0 && errno != EAGAIN)
{
D_NWSTACK("errno in UDPPort6::recvfrom: %s\n", strerror(errno));
return -1;
}
addr->setAddress(&sender);
#ifdef DEBUG_NW
char addrBuf[INET6_ADDRSTRLEN];
addr->sprint(addrBuf);
D_NWSTACK("sendto %s length = %d\n", addrBuf, status);
#endif
return status;
}

90
MQTTSNGateway/src/linux/udp6/SensorNetwork.h
View File

@@ -12,7 +12,7 @@
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* Contributors:
* Benjamin Aigner - port to UDPv6, used by RFC7668 (6lowpan over Bluetooth LE)
* Benjamin Aigner - port to UDPv6, used by RFC7668 (6lowpan over Bluetooth LE)
* Tomoaki Yamaguchi - initial API and implementation and/or initial documentation
**************************************************************************************/
@@ -22,69 +22,59 @@
#include "MQTTSNGWDefines.h"
#include <arpa/inet.h>
#include <string>
#include <poll.h>
using namespace std;
namespace MQTTSNGW
{
#ifdef DEBUG_NWSTACK
#define D_NWSTACK(...) printf(__VA_ARGS__)
#else
#define D_NWSTACK(...)
#endif
/*===========================================
Class SensorNetAddreess
============================================*/
class SensorNetAddress
{
public:
SensorNetAddress();
~SensorNetAddress();
void setAddress(struct sockaddr_in6 *IpAddr, uint16_t port);
int setAddress(string* data);
int setAddress(const char* data);
uint16_t getPortNo(void);
struct sockaddr_in6 *getIpAddress(void);
char* getAddress(void);
bool isMatch(SensorNetAddress* addr);
SensorNetAddress& operator =(SensorNetAddress& addr);
char* sprint(char* buf);
SensorNetAddress();
~SensorNetAddress();
void setAddress(sockaddr_in6 *IpAddr);
int setAddress(string* data);
int setAddress(const char* data);
uint16_t getPortNo(void);
sockaddr_in6* getIpAddress(void);
char* getAddress(void);
bool isMatch(SensorNetAddress* addr);
SensorNetAddress& operator =(SensorNetAddress& addr);
char* sprint(char* buf);
private:
uint16_t _portNo;
char _addrString[INET6_ADDRSTRLEN+1];
struct sockaddr_in6 _IpAddr;
char _addrString[INET6_ADDRSTRLEN + 1];
sockaddr_in6 _IpAddr;
};
/*========================================
Class UpdPort
Class UpdPort6
=======================================*/
class UDPPort6
{
public:
UDPPort6();
virtual ~UDPPort6();
UDPPort6();
virtual ~UDPPort6();
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);
int recv(uint8_t* buf, uint16_t len, SensorNetAddress* addr);
int open(uint16_t uniPortNo, uint16_t multiPortNo, const char *broadcastAddr, const char *interfaceName, uint32_t hops);
void close(void);
int unicast(const uint8_t* buf, uint32_t length, SensorNetAddress* sendToAddr);
int broadcast(const uint8_t* buf, uint32_t length);
int recv(uint8_t* buf, uint16_t len, SensorNetAddress* addr);
private:
void setNonBlocking(const bool);
int recvfrom(int sockfd, uint8_t* buf, uint16_t len, uint8_t flags, SensorNetAddress* addr);
void setNonBlocking(const bool);
int recvfrom(int sockfd, uint8_t* buf, uint16_t len, uint8_t flags, SensorNetAddress* addr);
int _sockfdUnicast;
int _sockfdMulticast;
char _interfaceName[10];
SensorNetAddress _grpAddr;
SensorNetAddress _clientAddr;
uint16_t _uniPortNo;
bool _disconReq;
unsigned int _hops;
pollfd _pollfds[2];
SensorNetAddress _grpAddr;
SensorNetAddress _clientAddr;
bool _disconReq;
uint32_t _hops;
};
/*===========================================
@@ -93,19 +83,19 @@ private:
class SensorNetwork: public UDPPort6
{
public:
SensorNetwork();
~SensorNetwork();
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);
void initialize(void);
const char* getDescription(void);
SensorNetAddress* getSenderAddress(void);
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);
void initialize(void);
const char* getDescription(void);
SensorNetAddress* getSenderAddress(void);
private:
SensorNetAddress _clientAddr; // Sender's address. not gateway's one.
string _description;
SensorNetAddress _clientAddr; // Sender's address. not gateway's one.
string _description;
};
}

7
MQTTSNGateway/src/linux/xbee/SensorNetwork.h
View File

@@ -25,13 +25,6 @@ using namespace std;
namespace MQTTSNGW
{
//#define DEBUG_NWSTACK
#ifdef DEBUG_NWSTACK
#define D_NWSTACK(...) printf(__VA_ARGS__)
#else
#define D_NWSTACK(...)
#endif
#define API_XMITREQUEST 0x10
#define API_RESPONSE 0x90

1
MQTTSNGateway/src/mainGateway.cpp
View File

@@ -21,7 +21,6 @@
#include "MQTTSNGWPacketHandleTask.h"
using namespace MQTTSNGW;
/*
* Gateway Application
*/