Dont exit on serial port failure

CMakeLists.txt

@@ -0,0 +1,12 @@
+cmake_minimum_required(VERSION 3.0)
+project(serialmultiplexer)
+
+set(SRC_FILES   main.cpp serial_io.cpp Socket.cpp )
+set(LIBS -pthread )
+
+add_executable(${PROJECT_NAME} ${SRC_FILES})
+
+target_link_libraries( ${PROJECT_NAME} ${LIBS})
+add_definitions(" -std=c++11 -Wall -O2 -g")
+
+install(TARGETS serialmultiplexer RUNTIME DESTINATION bin)

clienthandler.h

@@ -0,0 +1,125 @@
+#pragma once
+
+#include <vector>
+#include <array>
+#include <algorithm>
+#include <iostream>
+#include "Socket.h"
+#include "serial_io.h"
+
+class ClientHandler
+{
+private:
+    bool _isBroadcasting = false;
+    CommunicatingSocket *_socket;
+    bool _disconnected = false;
+    
+
+public:
+	
+    void cleanUp();
+    ClientHandler(CommunicatingSocket * const socket);
+    ~ClientHandler();
+    bool operator==(int fd);
+    bool operator!=(int fd);
+	bool operator==(ClientHandler& other);
+    bool operator!=(ClientHandler& other);
+    bool isDisconnected();
+    bool run(std::vector<ClientHandler>* clients, int serial, bool verbose = false);
+    void write(const char* buffer, const size_t len);
+	void dropData();
+};
+
+ClientHandler::ClientHandler(CommunicatingSocket * const socket): _socket(socket)
+{}
+
+ClientHandler::~ClientHandler()
+{
+}
+
+void ClientHandler::cleanUp()
+{
+	_socket->cleanUp();
+	delete  _socket;
+}
+
+void ClientHandler::dropData()
+{
+	std::array<char, 4096> buffer;
+	while(_socket->recv(buffer.data(), 4096) > 0);
+}
+
+bool ClientHandler::run(std::vector<ClientHandler>* clients, int serial, bool verbose)
+{
+    std::array<char, 4096> buffer;
+	int reclen = 0;
+    try 
+    {
+		reclen = _socket->recv(buffer.data(), 4096);
+		if(verbose) std::cout<<"Recived "<<reclen<<" bytes\n";
+	}
+    catch (SocketException &e)
+    {
+        std::cout<<e.what()<<std::endl;
+        _disconnected = true;
+    }
+    if(reclen > 0)
+    {
+        if(!_isBroadcasting && reclen >= 5 && strncmp( buffer.data(), "bcst:", 5) == 0) _isBroadcasting = true;
+        
+        if(_isBroadcasting) 
+        {
+			if(verbose) std::cout<<"Boradcasting "<<reclen<<" bytes\n";
+            for(ClientHandler& item : *clients) if(operator!=(item))item.write(buffer.data(), reclen);
+            char* newline = std::find(buffer.begin(), buffer.end(), '\n');
+            if(newline != std::end(buffer)) _isBroadcasting = false;
+        }
+        else
+        {
+			if(verbose) std::cout<<"wrote "<<reclen<<" bytes to serial\n";
+            if(sWrite(serial, buffer.data(), reclen) < 0 && (errno != EAGAIN || errno != EWOULDBLOCK)) 
+            {
+                throw serialIoException(serial, errno);
+            }
+        }
+    }
+    else if(reclen == 0) _disconnected = true; 
+    return !_disconnected;
+}
+
+bool ClientHandler::isDisconnected(){return _disconnected;}
+
+void ClientHandler::write(const char* buffer, const size_t len)
+{
+    try 
+    {
+		_socket->send(buffer, len);
+	}
+    catch (SocketException &e)
+    {
+        std::cout<<e.what()<<std::endl;
+        _disconnected = true;
+    }
+}
+
+bool ClientHandler::operator==(int fd)
+{
+    return fd == _socket->getFD();
+}
+
+
+bool ClientHandler::operator!=(int fd)
+{
+    return fd != _socket->getFD();
+}
+
+bool ClientHandler::operator==(ClientHandler& other)
+{
+    return _socket->getFD() == other._socket->getFD();
+}
+
+
+bool ClientHandler::operator!=(ClientHandler& other)
+{
+    return !(operator==(other));
+}

main.cpp

@@ -14,25 +14,17 @@
 
 #include "serial_io.h"
 #include "Socket.h"
+#include "clienthandler.h"
 
-#define VERSION "v0.5"
+#define VERSION "v0.7"
 
-volatile bool stop = false;
-volatile bool resettSerialPort = false;
-struct epoll_event ev;
+sig_atomic_t stop = false;
 
 void intHandler(int sig) 
 {
 	stop = true;
 }
 
-void alarmHandler(int sig)
-{
-	resettSerialPort = true;
-	signal(sig, alarmHandler);
-	alarm(600);
-}
-
 static void printUsage()
 {
 	std::cout<<"usage mulitplexer [option]\n\
@@ -85,7 +77,6 @@ static int parseCmdArgs(int argc, char** argv, Config *config)
 		{
 			config->noSerial=true;
 		}
-		
 		else if (std::string(argv[i]) == "-r" || std::string(argv[i]) == "--rates")
 		{
 			printRates();
@@ -103,18 +94,22 @@ static int parseCmdArgs(int argc, char** argv, Config *config)
 	return 0;
 }
 
-void acceptThreadFunction( TCPServerSocket* servSock,  std::vector<TCPSocket*>* clientSockets, std::mutex* clientSocketsMutex, int pollQue )
+void acceptThreadFunction( TCPServerSocket* servSock,  std::vector<ClientHandler>* clients, std::mutex* clientsMutex, int pollQue )
 {
 	while(!stop) 
 	{   
 		TCPSocket* newSock = servSock->accept();
 		if(newSock != nullptr)
 		{
-			clientSocketsMutex->lock();
-			clientSockets->push_back(newSock);       // Wait for a client to connect
-			clientSocketsMutex->unlock();
+			clientsMutex->lock();
+			clients->push_back(ClientHandler(newSock));       // Wait for a client to connect
+			struct epoll_event ev;
+			ev.events = EPOLLIN; 
+			ev.data.fd = newSock->getFD();
+			clientsMutex->unlock();
 			epoll_ctl(pollQue, EPOLL_CTL_ADD, newSock->getFD(), &ev);
-			clientSockets->back()->send("UVOS serial multiplexer " VERSION "\n", sizeof("UVOS serial multiplexer " VERSION "\n")-1);
+			char welcomeMesg[] =  "UVOS serial multiplexer " VERSION "\n";
+			clients->back().write(welcomeMesg, sizeof(welcomeMesg)-1);
 			std::cout<<"got client\n";
 		}
 		std::this_thread::sleep_for(std::chrono::seconds(1));
@@ -129,44 +124,73 @@ int openSerialPort(const Config& config)
 		std::cout<<"Opeing serial port: "<<config.portFileName<<" at "<<config.baud<<" baud\n";
 		serial = serialport_init(config.portFileName.c_str(), config.baud);
 		if(serial == -1) 
-		{
 			std::cout<<"Opeing serial port failed\n";
-			stop = true;
-		}
 		tcflush(serial, TCIOFLUSH);
 	}
 	else std::cout<<"Sinkless mode\n";
 	return serial;
 }
 
+int serialPortReconnect(Config& config, std::string err)
+{
+	std::cout<<"serial port execption "<<err<<"\ntrying to reopen\n";
+
+	sleep(10);
+	int serial = openSerialPort(config);
+	if(serial == -1)
+	{
+
+		if(config.portFileName == "/dev/ttyUSB0")
+		{
+			std::cout<<"Could not reopen "<<config.portFileName<<" trying /dev/ttyUSB1\n";
+			config.portFileName = "/dev/ttyUSB1";
+			serial = openSerialPort(config);
+		}
+		else if(config.portFileName == "/dev/ttyUSB1")
+		{
+			std::cout<<"Could not reopen "<<config.portFileName<<" trying /dev/ttyUSB0\n";
+			config.portFileName = "/dev/ttyUSB0";
+			serial = openSerialPort(config);
+		}
+
+		if(serial == -1)
+			std::cout<<"Could not reopen "<<config.portFileName<<'\n';
+	}
+	return serial;
+}
+
 int main(int argc, char* argv[])
 {
 	Config config;
 	
-	if(parseCmdArgs(argc, argv, &config) != 0) return -1;
+	if(parseCmdArgs(argc, argv, &config) != 0)
+		return -1;
 	
 	std::cout<<"UVOS serial mulitplexer "<<VERSION<<'\n';
 	
 	int pollQue = epoll_create1(0);
-	ev.events = EPOLLIN; 
+
+	std::mutex clientsMutex;
+	std::vector<ClientHandler> clients;
 
 	int serial = openSerialPort(config);
 	if(!config.noSerial)
 	{
-		std::cout<<"Using serial port: "<<config.portFileName<<" at "<<config.baud<<" baud\n";
-		serial = serialport_init(config.portFileName.c_str(), config.baud);
-		if(serial == -1) return 1;
+		if(serial == -1)
+			return 1;
 		else 
 		{
-			struct epoll_event ev;
+			struct epoll_event ev = {};
+			ev.events = EPOLLIN; 
+			ev.data.fd = -1;
 			epoll_ctl(pollQue, EPOLL_CTL_ADD, serial, &ev);
 		}
 		tcflush(serial, TCIOFLUSH);
 	}
-	else std::cout<<"Sinkless mode\n";
-	
-	std::mutex clientSocketsMutex;
-	std::vector<TCPSocket*> clientSockets;
+	else 
+	{
+		std::cout<<"Sinkless mode\n";
+	}
 	
 	std::thread* acceptThread;
 	TCPServerSocket* servSock;
@@ -174,9 +198,9 @@ int main(int argc, char* argv[])
 	std::cout<<"opening TCP socket on port "<<config.port<<'\n';
 	try
 	{
-		servSock = new TCPServerSocket(config.port, 5, true);     // Server Socket object
+		servSock = new TCPServerSocket(config.port, 5, true);
 		servSock->setBlocking(false);
-		acceptThread = new std::thread(acceptThreadFunction, servSock, &clientSockets, &clientSocketsMutex, pollQue);
+		acceptThread = new std::thread(acceptThreadFunction, servSock, &clients, &clientsMutex, pollQue);
 	}
 	catch(SocketException &e)
 	{
@@ -184,114 +208,80 @@ int main(int argc, char* argv[])
 		return 1;
 	}
 	
-	char buffer[4096];
-	
 	signal(SIGINT, intHandler);
 	signal(SIGTERM, intHandler);
-	signal(SIGALRM, alarmHandler);
-	
-	if(config.reinit) alarm(600);
-	
+    signal(SIGPIPE, SIG_IGN);
 	
 	std::cout<<"starting loop\n";
 	
 	while(!stop)
 	{
-		epoll_wait(pollQue, &ev, 1, 2000);
-		int readlen = sRead(serial, buffer, 4096);
-		for(unsigned int i = 0; i < clientSockets.size(); i++)
+		struct epoll_event ev;
+		if(epoll_wait(pollQue, &ev, 1, 2000) == 1)
+		{
+			if(ev.data.fd != -1)
+			{
+				std::vector<ClientHandler>::iterator client =
+					std::find(clients.begin(), clients.end(), ev.data.fd);
+				if(client == clients.end())
+					continue;
+				clientsMutex.lock();
+				std::cout<<"client poll\n";
+				if(ev.events & EPOLLIN)
 				{
-			clientSocketsMutex.lock();
 					try
 					{
-				if(readlen > 0) 
-				{
-					if(config.verbose) 
-					{
-						std::cout<<"bcst: ";
-						for( int j = 0; j < readlen; j++  )std::cout<<buffer[j];
-						std::cout<<std::endl;
+						client->run(&clients, serial, config.verbose);
 					}
-					clientSockets[i]->send(buffer, readlen);
-				}
-				
-				char inBuffer[4096];
-				int reclen = clientSockets[i]->recv(inBuffer, 4096);
-				if( reclen > 0 ) 
-				{
-					if(config.verbose)
-					{
-						std::cout<<"rec: ";
-						for( int j = 0; j < reclen; j++  )std::cout<<inBuffer[j];
-					}
-					if(strncmp( inBuffer, "bcst:", 5) == 0)
-					{
-						if(config.verbose) 
-						{
-							std::cout<<"bcst: ";
-							for( int j = 6; j < reclen; j++  )std::cout<<inBuffer[j];
-							std::cout<<std::endl;
-						}
-						for(unsigned int j = 0; j < clientSockets.size(); j++) if(i != j) clientSockets[j]->send(inBuffer+6, reclen-6);
-						
-					}
-					else if(strncmp( inBuffer, "reinitsp", 8) == 0 && serial != -1)
-					{
-						if(serialport_set_config(serial, config.baud) == 0)
-						{
-							char response[] = "reinit serial port succsesfull\n";
-							std::cerr<<response;
-							clientSockets[i]->send(response, sizeof(response));
-						}
-						else
-						{
-							char response[] = "reinit serial port failed!\n";
-							std::cerr<<response;
-							clientSockets[i]->send(response, sizeof(response));
-						}
-					}
-					else
-					{
-						if(config.verbose) 
-						{
-							std::cout<<"wrote \"";
-							for( int j = 0; j < reclen; j++  )std::cout<<inBuffer[j];
-							std::cout<<"\" to serial \n";
-						}
-						sWrite(serial, inBuffer, reclen);
-					}
-					if(config.verbose) std::cout<<std::endl;
-				}
-				else if(reclen == 0)
-				{
-					std::cout<<"client "<<i<<" disconnected"<<'\n';
-					clientSockets[i]->cleanUp();
-					clientSockets.erase(clientSockets.begin()+i);
-					i--;
-					if(i < 0) i=0;
-				}
-			}
-			catch(SocketException &e)
-			{
-				std::cout<<e.what()<<std::endl;
-				clientSockets[i]->cleanUp();
-				clientSockets.erase(clientSockets.begin()+i);
-				i--;
-				if(i < 0) i=0;
-			}
-			clientSocketsMutex.unlock();
-		}
-		if(resettSerialPort)
+					catch(serialIoException& ex)
 					{
 						close(serial);
-			serial = openSerialPort(config);
-			resettSerialPort = false;
+						serialPortReconnect(config, ex.what());
+					}
+				}
+				if((ev.events & (EPOLLHUP | EPOLLERR)) || client->isDisconnected()) 
+				{
+					client->cleanUp();
+					clients.erase(client);
+					std::cout<<"client disconnected\n";
+				}
+				clientsMutex.unlock();
+			}
+			else
+			{
+				char buffer[4096];
+				ssize_t readlen = sRead(serial, buffer, 4096);
+				if(readlen < 0 && (errno != EAGAIN || errno != EWOULDBLOCK)) 
+				{
+					close(serial);
+					serialPortReconnect(config, strerror(errno));
+				}
+				if(config.verbose)
+				{
+					std::cout<<"Sending \"";
+					for(ssize_t i = 0; i < readlen; ++i)
+					{
+						if(buffer[i] == '\n')
+							std::cout<<"\\n";
+						else
+							std::cout<<buffer[i];
+					}
+					std::cout<<"\" to clients from serial\n";
+				}
+				clientsMutex.lock();
+				for(ClientHandler& client : clients) 
+				{
+					client.write(buffer, readlen);
+				}
+				clientsMutex.unlock();
+			}
 		}
 	}
 	
 	acceptThread->join();
 	delete acceptThread;
-	for(unsigned int i = 0; i < clientSockets.size(); i++) clientSockets[i]->cleanUp();
+	for(ClientHandler& client : clients)
+		client.cleanUp();
 	servSock->cleanUp();
 	delete servSock;
 	return 0;

serial_io.cpp

@@ -1,13 +1,15 @@
 #include "serial_io.h"
 
-void sWrite(int port, char string[], size_t length)
+ssize_t sWrite(int port, char string[], size_t length)
 {
-    if(port != -1) write(port, string, length);
+    if(port != -1) return write(port, string, length);
+    else return 0;
 }
 
-void sWrite(int port, const char string[], size_t length)
+ssize_t sWrite(int port, const char string[], size_t length)
 {
-    if(port != -1) write(port, string, length);
+    if(port != -1) return write(port, string, length);
+    else return 0;
 }
 
 ssize_t sRead(int port, void *buf, size_t count)

serial_io.h

@@ -6,13 +6,15 @@
 
 #ifdef __cplusplus
 #include <iostream>
+#include <stdexcept>
+#include <string.h>
 #endif
 
 #define BAUDRATE B38400
 
-void sWrite(int port, char string[], size_t length);
+ssize_t sWrite(int port, char string[], size_t length);
 
-void sWrite(int port, const char string[], size_t length);
+ssize_t sWrite(int port, const char string[], size_t length);
 
 ssize_t sRead(int port, void *buf, size_t count);
 
@@ -24,5 +26,17 @@ int serialport_set_config(int fd, int baud);
 
 int serialport_init(const char* device, int baud = BAUDRATE, bool block = false);
 
+#ifdef __cplusplus
+class serialIoException: public std::runtime_error
+{
+    public:
+        int fd;
+        int errorNumber;
+        serialIoException(int fd_, int errorNumber_):
+        std::runtime_error("file descriptor error, fd: " + std::to_string(fd_) + " error: " + strerror(errorNumber_) + "\n"), fd(fd_), errorNumber(errorNumber_)
+        {}
+};
+#endif
+
 
 #endif // SERIAL_H