thooge
/
esp32-nmea2000-obp60
mirror of https://github.com/thooge/esp32-nmea2000-obp60.git
1
0
Fork 0
Code Releases Activity

get rid of wifi client, directly use the socket api

This commit is contained in:
wellenvogel 2021-12-28 18:08:06 +01:00
parent fb435499a1
commit 61e0e66acb
2 changed files with 397 additions and 191 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

507
lib/socketserver/GwSocketServer.cpp
View File

@ -3,188 +3,379 @@
#include <lwip/sockets.h> #include <lwip/sockets.h>
#include "GwBuffer.h" #include "GwBuffer.h"
class GwClient{ class GwClient
public: {
wiFiClientPtr client; public:
int fd=-1;
int overflows; int overflows;
String remoteIp; String remoteIpAddress;
private:
unsigned long lastWrite=0; private:
unsigned long writeTimeout=10000; unsigned long lastWrite = 0;
bool pendingWrite=false; unsigned long writeTimeout = 10000;
bool writeError=false; bool pendingWrite = false;
bool writeError = false;
bool allowRead; bool allowRead;
GwBuffer *buffer=NULL; GwBuffer *buffer = NULL;
GwBuffer *readBuffer=NULL; GwBuffer *readBuffer = NULL;
GwLog *logger; GwLog *logger;
public:
GwClient(wiFiClientPtr client,GwLog *logger,int id, bool allowRead=false){ public:
this->client=client; static IPAddress remoteIP(int fd)
this->logger=logger; {
this->allowRead=allowRead; struct sockaddr_storage addr;
String bufName="Sock("; socklen_t len = sizeof addr;
bufName+=String(id); getpeername(fd, (struct sockaddr*)&addr, &len);
bufName+=")"; struct sockaddr_in *s = (struct sockaddr_in *)&addr;
buffer=new GwBuffer(logger,GwBuffer::TX_BUFFER_SIZE,bufName+"wr"); return IPAddress((uint32_t)(s->sin_addr.s_addr));
if (allowRead){
readBuffer=new GwBuffer(logger,GwBuffer::RX_BUFFER_SIZE,bufName+"rd");
} }
overflows=0; GwClient(GwLog *logger, int id, bool allowRead = false)
if (client != NULL){ {
remoteIp=client->remoteIP().toString(); this->logger = logger;
this->allowRead = allowRead;
String bufName = "Sock(";
bufName += String(id);
bufName += ")";
buffer = new GwBuffer(logger, GwBuffer::TX_BUFFER_SIZE, bufName + "wr");
if (allowRead)
{
readBuffer = new GwBuffer(logger, GwBuffer::RX_BUFFER_SIZE, bufName + "rd");
} }
overflows = 0;
} }
void setClient(wiFiClientPtr client){ void setClient(int fd)
this->client=client; {
this->fd = fd;
buffer->reset("new client"); buffer->reset("new client");
if (readBuffer) readBuffer->reset("new client"); if (readBuffer)
overflows=0; readBuffer->reset("new client");
pendingWrite=false; overflows = 0;
writeError=false; pendingWrite = false;
lastWrite=0; writeError = false;
if (client){ lastWrite = 0;
remoteIp=client->remoteIP().toString(); if (fd >= 0)
{
remoteIpAddress = remoteIP(fd).toString();
} }
else{ else
remoteIp=String("---"); {
remoteIpAddress = String("---");
} }
} }
bool hasClient(){ bool hasClient()
return client != NULL; {
return fd >= 0;
} }
~GwClient(){ void stop(){
if (fd >= 0){
close(fd);
fd=-1;
}
}
~GwClient()
{
delete buffer; delete buffer;
if (readBuffer) delete readBuffer; if (readBuffer)
delete readBuffer;
} }
bool enqueue(uint8_t *data, size_t len){ bool connected()
if (len == 0) return true; {
size_t rt=buffer->addData(data,len); if (fd >= 0)
if (rt < len){ {
LOG_DEBUG(GwLog::LOG,"overflow on %s",remoteIp.c_str()); uint8_t dummy;
int res = recv(fd, &dummy, 0, MSG_DONTWAIT);
// avoid unused var warning by gcc
(void)res;
// recv only sets errno if res is <= 0
if (res <= 0)
{
switch (errno)
{
case EWOULDBLOCK:
case ENOENT: //caused by vfs
return true;
break;
case ENOTCONN:
case EPIPE:
case ECONNRESET:
case ECONNREFUSED:
case ECONNABORTED:
return false;
break;
default:
return true;
}
}
else
{
return true;
}
}
return false;
}
bool enqueue(uint8_t *data, size_t len)
{
if (len == 0)
return true;
size_t rt = buffer->addData(data, len);
if (rt < len)
{
LOG_DEBUG(GwLog::LOG, "overflow on %s", remoteIpAddress.c_str());
overflows++; overflows++;
return false; return false;
} }
return true; return true;
} }
bool hasData(){ bool hasData()
{
return buffer->usedSpace() > 0; return buffer->usedSpace() > 0;
} }
bool handleError(int res,bool errorIf0=true){ bool handleError(int res, bool errorIf0 = true)
if (res == 0 && errorIf0){ {
LOG_DEBUG(GwLog::LOG,"client shutdown (recv 0) on %s",remoteIp.c_str()); if (res == 0 && errorIf0)
client->stop(); {
LOG_DEBUG(GwLog::LOG, "client shutdown (recv 0) on %s", remoteIpAddress.c_str());
stop();
return false; return false;
} }
if (res < 0){ if (res < 0)
if (errno != EAGAIN){ {
LOG_DEBUG(GwLog::LOG,"client read error %d on %s",errno,remoteIp.c_str()); if (errno != EAGAIN)
client->stop(); {
LOG_DEBUG(GwLog::LOG, "client read error %d on %s", errno, remoteIpAddress.c_str());
stop();
return false; return false;
} }
return false; return false;
} }
return true; return true;
} }
GwBuffer::WriteStatus write(){ GwBuffer::WriteStatus write()
if (! hasClient()) { {
LOG_DEBUG(GwLog::LOG,"write called on empty client"); if (!hasClient())
{
LOG_DEBUG(GwLog::LOG, "write called on empty client");
return GwBuffer::ERROR; return GwBuffer::ERROR;
} }
if (! buffer->usedSpace()){ if (!buffer->usedSpace())
pendingWrite=false; {
pendingWrite = false;
return GwBuffer::OK; return GwBuffer::OK;
} }
buffer->fetchData(-1,[](uint8_t *buffer, size_t len, void *param)->size_t{ buffer->fetchData(
GwClient *c=(GwClient*)param; -1, [](uint8_t *buffer, size_t len, void *param) -> size_t
int res = send(c->client->fd(), (void*) buffer, len, MSG_DONTWAIT); {
if (! c->handleError(res,false)) return 0; GwClient *c = (GwClient *)param;
if (res >= len){ int res = send(c->fd, (void *)buffer, len, MSG_DONTWAIT);
c->pendingWrite=false; if (!c->handleError(res, false))
return 0;
if (res >= len)
{
c->pendingWrite = false;
} }
else{ else
if (!c->pendingWrite){ {
c->lastWrite=millis(); if (!c->pendingWrite)
c->pendingWrite=true; {
c->lastWrite = millis();
c->pendingWrite = true;
} }
else{ else
{
//we need to check if we have still not been able //we need to check if we have still not been able
//to write until timeout //to write until timeout
if (millis() >= (c->lastWrite+c->writeTimeout)){ if (millis() >= (c->lastWrite + c->writeTimeout))
c->logger->logDebug(GwLog::ERROR,"Write timeout on channel %s",c->remoteIp.c_str()); {
c->writeError=true; c->logger->logDebug(GwLog::ERROR, "Write timeout on channel %s", c->remoteIpAddress.c_str());
c->writeError = true;
} }
} }
} }
return res; return res;
},this); },
if (writeError){ this);
LOG_DEBUG(GwLog::DEBUG+1,"write error on %s",remoteIp.c_str()); if (writeError)
{
LOG_DEBUG(GwLog::DEBUG + 1, "write error on %s", remoteIpAddress.c_str());
return GwBuffer::ERROR; return GwBuffer::ERROR;
} }
return GwBuffer::OK; return GwBuffer::OK;
} }
bool read(){ bool read()
if (! allowRead){ {
size_t maxLen=100; if (!allowRead)
{
size_t maxLen = 100;
char buffer[maxLen]; char buffer[maxLen];
int res = recv(client->fd(), (void*) buffer, maxLen, MSG_DONTWAIT); int res = recv(fd, (void *)buffer, maxLen, MSG_DONTWAIT);
return handleError(res); return handleError(res);
} }
readBuffer->fillData(-1,[](uint8_t *buffer, size_t len, void *param)->size_t{ readBuffer->fillData(
GwClient *c=(GwClient*)param; -1, [](uint8_t *buffer, size_t len, void *param) -> size_t
int res = recv(c->client->fd(), (void*) buffer, len, MSG_DONTWAIT); {
if (! c->handleError(res)) return 0; GwClient *c = (GwClient *)param;
int res = recv(c->fd, (void *)buffer, len, MSG_DONTWAIT);
if (!c->handleError(res))
return 0;
return res; return res;
},this); },
this);
return true; return true;
} }
bool messagesFromBuffer(GwMessageFetcher *writer){ bool messagesFromBuffer(GwMessageFetcher *writer)
if (! allowRead) return false; {
if (!allowRead)
return false;
return writer->handleBuffer(readBuffer); return writer->handleBuffer(readBuffer);
} }
}; };
class GwTcpClient
GwSocketServer::GwSocketServer(const GwConfigHandler *config,GwLog *logger,int minId){
this->config=config;
this->logger=logger;
this->minId=minId;
maxClients=1;
allowReceive=false;
}
void GwSocketServer::begin(){
maxClients=config->getInt(config->maxClients);
allowReceive=config->getBool(config->readTCP);
clients=new gwClientPtr[maxClients];
for (int i=0;i<maxClients;i++){
clients[i]=gwClientPtr(new GwClient(wiFiClientPtr(NULL),logger,i,allowReceive));
}
server=new WiFiServer(config->getInt(config->serverPort),maxClients+1);
server->begin();
LOG_DEBUG(GwLog::LOG,"Socket server created, port=%d",
config->getInt(config->serverPort));
MDNS.addService("_nmea-0183","_tcp",config->getInt(config->serverPort));
}
void GwSocketServer::loop(bool handleRead,bool handleWrite)
{ {
if (! clients) return; GwClient *gwClient = NULL;
WiFiClient client = server->available(); // listen for incoming clients IPAddress remoteAddress;
uint16_t port = 0;
GwLog *logger;
if (client) public:
typedef enum
{ {
LOG_DEBUG(GwLog::LOG,"new client connected from %s", C_DISABLED = 0,
client.remoteIP().toString().c_str()); C_INITIALIZED = 1,
fcntl(client.fd(), F_SETFL, O_NONBLOCK); C_CONNECTING = 2,
C_CONNECTED = 3
} State;
private:
State state = C_DISABLED;
void stop()
{
if (gwClient->hasClient())
{
LOG_DEBUG(GwLog::DEBUG, "stopping tcp client");
gwClient->stop();
}
state = C_DISABLED;
}
void startConnection()
{
//TODO
state = C_CONNECTING;
}
void checkConnection()
{
}
public:
GwTcpClient(GwLog *logger, GwClient *gwClient)
{
this->logger = logger;
this->gwClient = gwClient;
}
void begin(IPAddress address, uint16_t port)
{
stop();
this->remoteAddress = address;
this->port = port;
state = C_INITIALIZED;
startConnection();
}
void loop()
{
if (state == C_CONNECTING)
{
checkConnection();
}
}
};
GwSocketServer::GwSocketServer(const GwConfigHandler *config, GwLog *logger, int minId)
{
this->config = config;
this->logger = logger;
this->minId = minId;
maxClients = 1;
allowReceive = false;
}
bool GwSocketServer::createListener()
{
struct sockaddr_in server;
listener = socket(AF_INET, SOCK_STREAM, 0);
if (listener < 0)
{
return false;
}
int enable = 1;
setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(int));
server.sin_family = AF_INET;
server.sin_addr.s_addr = INADDR_ANY;
server.sin_port = htons(listenerPort);
if (bind(listener, (struct sockaddr *)&server, sizeof(server)) < 0)
return false;
if (listen(listener, maxClients) < 0)
return false;
fcntl(listener, F_SETFL, O_NONBLOCK);
return true;
}
void GwSocketServer::begin()
{
maxClients = config->getInt(config->maxClients);
allowReceive = config->getBool(config->readTCP);
listenerPort=config->getInt(config->serverPort);
clients = new GwClient*[maxClients];
for (int i = 0; i < maxClients; i++)
{
clients[i] = new GwClient(logger, i, allowReceive);
}
if (! createListener()){
listener=-1;
LOG_DEBUG(GwLog::ERROR,"Unable to create listener");
return;
}
LOG_DEBUG(GwLog::LOG, "Socket server created, port=%d",
config->getInt(config->serverPort));
MDNS.addService("_nmea-0183", "_tcp", config->getInt(config->serverPort));
}
int GwSocketServer::available()
{
if (listener < 0)
return -1;
int client_sock;
struct sockaddr_in _client;
int cs = sizeof(struct sockaddr_in);
client_sock = lwip_accept_r(listener, (struct sockaddr *)&_client, (socklen_t *)&cs);
if (client_sock >= 0)
{
int val = 1;
if (setsockopt(client_sock, SOL_SOCKET, SO_KEEPALIVE, (char *)&val, sizeof(int)) == ESP_OK)
{
if (setsockopt(client_sock, IPPROTO_TCP, TCP_NODELAY, (char *)&val, sizeof(int)) == ESP_OK)
fcntl(client_sock, F_SETFL, O_NONBLOCK);
return client_sock;
}
close(client_sock);
}
return -1;
}
void GwSocketServer::loop(bool handleRead, bool handleWrite)
{
if (!clients)
return;
int client = available(); // listen for incoming clients
if (client >= 0)
{
LOG_DEBUG(GwLog::LOG, "new client connected from %s",
GwClient::remoteIP(client).toString().c_str());
bool canHandle = false; bool canHandle = false;
for (int i = 0; i < maxClients; i++) for (int i = 0; i < maxClients; i++)
{ {
if (!clients[i]->hasClient()) if (!clients[i]->hasClient())
{ {
clients[i]->setClient(wiFiClientPtr(new WiFiClient(client))); clients[i]->setClient(client);
LOG_DEBUG(GwLog::LOG,"set client as number %d", i); LOG_DEBUG(GwLog::LOG, "set client as number %d", i);
canHandle = true; canHandle = true;
break; break;
} }
@ -192,7 +383,7 @@ void GwSocketServer::loop(bool handleRead,bool handleWrite)
if (!canHandle) if (!canHandle)
{ {
logger->logDebug(GwLog::ERROR, "no space to store client, disconnect"); logger->logDebug(GwLog::ERROR, "no space to store client, disconnect");
client.stop(); close(client);
} }
} }
if (handleWrite) if (handleWrite)
@ -200,69 +391,83 @@ void GwSocketServer::loop(bool handleRead,bool handleWrite)
//sending //sending
for (int i = 0; i < maxClients; i++) for (int i = 0; i < maxClients; i++)
{ {
gwClientPtr client = clients[i]; GwClient *client = clients[i];
if (!client->hasClient()) if (!client->hasClient())
continue; continue;
GwBuffer::WriteStatus rt = client->write(); GwBuffer::WriteStatus rt = client->write();
if (rt == GwBuffer::ERROR) if (rt == GwBuffer::ERROR)
{ {
LOG_DEBUG(GwLog::ERROR, "write error on %s, closing", client->remoteIp.c_str()); LOG_DEBUG(GwLog::ERROR, "write error on %s, closing", client->remoteIpAddress.c_str());
client->client->stop(); client->stop();
} }
} }
} }
for (int i = 0; i < maxClients; i++) for (int i = 0; i < maxClients; i++)
{ {
gwClientPtr client = clients[i]; GwClient *client = clients[i];
if (!client->hasClient()) if (!client->hasClient())
continue; continue;
if (!client->client->connected()) if (!client->connected())
{ {
LOG_DEBUG(GwLog::LOG,"client %d disconnect %s", i, client->remoteIp.c_str()); LOG_DEBUG(GwLog::LOG, "client %d disconnect %s", i, client->remoteIpAddress.c_str());
client->client->stop(); client->stop();
client->setClient(NULL);
} }
else else
{ {
if (handleRead) client->read(); if (handleRead)
client->read();
} }
} }
} }
bool GwSocketServer::readMessages(GwMessageFetcher *writer){ bool GwSocketServer::readMessages(GwMessageFetcher *writer)
if (! allowReceive || ! clients) return false; {
bool hasMessages=false; if (!allowReceive || !clients)
for (int i = 0; i < maxClients; i++){ return false;
writer->id=minId+i; bool hasMessages = false;
if (!clients[i]->hasClient()) continue; for (int i = 0; i < maxClients; i++)
if (clients[i]->messagesFromBuffer(writer)) hasMessages=true; {
writer->id = minId + i;
if (!clients[i]->hasClient())
continue;
if (clients[i]->messagesFromBuffer(writer))
hasMessages = true;
} }
return hasMessages; return hasMessages;
} }
void GwSocketServer::sendToClients(const char *buf,int source){ void GwSocketServer::sendToClients(const char *buf, int source)
if (! clients) return; {
int len=strlen(buf); if (!clients)
int sourceIndex=source-minId; return;
int len = strlen(buf);
int sourceIndex = source - minId;
for (int i = 0; i < maxClients; i++) for (int i = 0; i < maxClients; i++)
{ {
if (i == sourceIndex)continue; //never send out to the source we received from if (i == sourceIndex)
gwClientPtr client = clients[i]; continue; //never send out to the source we received from
if (! client->hasClient()) continue; GwClient *client = clients[i];
if ( client->client->connected() ) { if (!client->hasClient())
client->enqueue((uint8_t*)buf,len); continue;
if (client->connected())
{
client->enqueue((uint8_t *)buf, len);
} }
} }
} }
int GwSocketServer::numClients(){ int GwSocketServer::numClients()
if (! clients) return 0; {
int num=0; if (!clients)
for (int i = 0; i < maxClients; i++){ return 0;
if (clients[i]->hasClient()) num++; int num = 0;
for (int i = 0; i < maxClients; i++)
{
if (clients[i]->hasClient())
num++;
} }
return num; return num;
} }
GwSocketServer::~GwSocketServer(){ GwSocketServer::~GwSocketServer()
{
} }

9
lib/socketserver/GwSocketServer.h
View File

@ -6,18 +6,19 @@
#include <memory> #include <memory>
#include <WiFi.h> #include <WiFi.h>
using wiFiClientPtr = std::shared_ptr<WiFiClient>;
class GwClient; class GwClient;
using gwClientPtr = std::shared_ptr<GwClient>;
class GwSocketServer{ class GwSocketServer{
private: private:
const GwConfigHandler *config; const GwConfigHandler *config;
GwLog *logger; GwLog *logger;
gwClientPtr *clients=NULL; GwClient **clients=NULL;
WiFiServer *server=NULL; int listener=-1;
int listenerPort=-1;
bool allowReceive; bool allowReceive;
int maxClients; int maxClients;
int minId; int minId;
bool createListener();
int available();
public: public:
GwSocketServer(const GwConfigHandler *config,GwLog *logger,int minId); GwSocketServer(const GwConfigHandler *config,GwLog *logger,int minId);
~GwSocketServer(); ~GwSocketServer();