Files
1NiM-WatingRoom-WatchContro…/components/spit/src/spit_client.cpp
T
2026-01-05 19:53:30 +01:00

650 lines
24 KiB
C++

#include "spit_client.hpp"
#include "esp_log.h"
#include "Queue.hpp"
namespace spit {
static const char *TAG = "SPIT";
static fos::Queue<vector<char>> incommingSpit(128);
/**
* Create a new SPIT_client instance, should be done once by the user app.
* @param p_SPITClientInterface the user app should implement the SPIT_ClientInterface.
*/
SClient::SClient(string name, string projName, ISocket &socket, Callbacks calls)
: ISadClient(),
mCallbacks(calls),
mProcessor(name, projName, getClientHostName(), createGUID(), calls,
{
.delay = [&](int ms){ return fos::Task::delayMs(ms); },
.sendMessage = [&](Message::OPPart &msg){ return sendMessage(msg); },
.parseMessage = [&](Message& msg) { parseMessage(msg); }
}
),
mSocket(socket) {
}
SClient::~SClient() {
stopServerAutoDetect();
disconnectInternal(Connection::None);
}
void SClient::clientProjectClose() {
if (mConnectionStatus == Connection::Ok) {
ESP_LOGI(TAG, " ... tidy up ressources");
Message::CloseProject closeSPITProject(mProcessor.getProjName());
if (sendMessage(closeSPITProject) < 0) {
ESP_LOGD(TAG, "Client could not send Client's ReportProject");
}
}
}
void SClient::clientProjectLoaded() {
// set confirmed = false for all SPITTypes and SPITObjects
mProcessor.resetConfirmedFlags();
if (mConnectionStatus == Connection::Ok) {
// send ReportProject to the server
Message::ReportProject reportSPITProject(mProcessor.getProjName(), mProcessor.getProjName());
reportSPITProject.setFramesPerSecond(0);
if (sendMessage(reportSPITProject) < 0) {
ESP_LOGD(TAG, "SPIT_CLient::clientProjectLoaded could not send Client's ReportProject");
}
// send RequestProjectTransfer
Message::RequestProjectTransfer sMRequestProjectTransfer;
if (sendMessage(sMRequestProjectTransfer) < 0) {
ESP_LOGD(TAG, "SPIT_CLient::clientProjectLoaded could not send RequestProjectTransfer");
}
}
}
//---------- ServerAutoDetection SAD -----------
/** Starts the autodetection of SPITServers in the local network. If a connection is tried, the autodetection should be stopped before.
* If a SPITServer is detected in the locale network an eventServerDetected occurs.
*/
void SClient::startServerAutoDetect() {
if (mDisconnectedByServer == true) {
// Server has previously disconnected -> do not start ServerAutodetect directly
fos::Task::delayMs(1000); // sleep a little to give the SPITServer time to clean up its stuff
}
SadDetectedServer::clearDetectedServers();
if (!sThread) {
sThread = new fos::TaskT<SClient &>(SClient::processSocketTask, *this, "spitSock", fos::Task::DefaultStackSize * 2);
}
mAbortSearch = false;
sStartSad.give();
}
/** Stops the auto detection of SPIT servers. */
void SClient::stopServerAutoDetect() {
mAbortSearch = true;
}
/**
* If a new server was found by the ServerAutoDetect mechanism, this function is called.
* To do so the SAD_Client ServerAutoDetect has to be startet - see SAD_Client::startServerAutoDetect.
* If a server can be reached via several IP addresses, the IP address is selected according to the following preference:
*/
void SClient::eventServerDetected(SadDetectedServer &server) {
mCallbacks.onServerEvent(ServerEvent::Detect, server);
// If a connection has been established and the connection is lost -> reconnect
if (_doReconnect) {
if (mServerID == server.getID()) {
stopServerAutoDetect();
ESP_LOGI(TAG, "reconnect to %s / %s", server.getIP().data(), server.getPort().data());
if (connect(server.getIP(), server.getPort(), server.getClientIP(), "0")) {
return;
}
}
}
// Stop auto detection because a suitable server was found
stopServerAutoDetect();
ESP_LOGI(TAG, "connect to %s / %s", server.getIP().data(), server.getPort().data());
if (!connect(server.getIP(), server.getPort(), server.getClientIP(), "0" /*= any port*/)) {
// connection message could not be sent -> better to restart AutoDetection
startServerAutoDetect();
}
}
/**
* If a SPITServer is not more found, this function is called.
* To do so the SAD_Client ServerAutoDetect has to be startet - see SAD_Client::startServerAutoDetect.
*/
void SClient::eventServerLost(SadDetectedServer &server) {
mCallbacks.onServerEvent(ServerEvent::Lost, server);
}
void SClient::eventServerChanged(SadDetectedServer &server) {
mCallbacks.onServerEvent(ServerEvent::Changed, server);
// tryConnection(detectedServer);
}
// void MyApp::tryConnection(const SadDetectedServer &detected) {
// // fetch the infoemations from the detected
// bool boolSuccess = false;
// bool boolPossibleConnect = checkPossibleConnection(detected);
// if (boolPossibleConnect == false)
// return;
// // --------------------------------------
// // You have three Options to use the detected server
// // OPTION 1 Use the first server found
// // OPTION 2 Use the first server found that fulfills specific conditions
// // OPTION 3 take a copy of the detected and do connection later
// // SadDetectedServer o_DetectedServer = detected; //this will call the copy constructor
// // ---------------------------------------
// // Excample use the first detected LOOPBACK server, means the server runs on the same computer as this client
// // if (detected.getIpIsLoopback() == true) { //omit the if condition if you want to use the first server found
// boolSuccess = false;
// // Stop auto detection because a suitable server was found
// stopServerAutoDetect();
// if (mSPIT_Client != nullptr) {
// mServerProjectName = detected.getProjectName();
// mServerIP = detected.getIP();
// #if defined(DEBUGOUTPUT)
// ESP_LOGD(TAG, "MySPITApp: connect to %s ", detected.getIP().data());
// #endif
// // Use the mBoardNumber as login name
// string stringLoginName = "";
// string stringLoginPassword = "";
// boolSuccess = mSPIT_Client->connect(detected.getIP(), detected.getPort(), detected.getClientIP(), stringLoginName, stringLoginPassword); // ClientPort = "0" meeans use any port
// }
// if (boolSuccess == false) {
// // connection message could not be sent
// // perhaps restart AutoDetection
// startServerAutoDetect();
// } else {
// // WAITING FOR CONNECTION ANSWER
// // If the Server is closed in the meantime no answer will return
// //-> connect(..) has started the watchdog
// }
// // } //Excample
// }
int SClient::sendMessage(Message::OPPart &opPart) {
if (mMessageNumberSend == 255)
mMessageNumberSend = 1;
else
mMessageNumberSend++;
opPart.setMessageNumber(mMessageNumberSend);
opPart.setServerInfoInHeader(mProcessor.getID(), mServerID);
int opFlag = opPart.getHeader()->getType();
try {
auto spitMessage = Message::createMessage(opPart);
if (spitMessage.empty()) {
return -1;
}
int intResult = mSocket.sendMessage(spitMessage.data(), spitMessage.size());
if (intResult < 0) {
ESP_LOGD(TAG, "client could not send message OPFlage: %d", opFlag);
}
return intResult;
} catch (...) {
ESP_LOGD(TAG, "Exception client OPFlag: %d", opFlag);
}
return -1;
}
bool SClient::startReceiveSPITMessages() {
if (!mSocket.getIsBound())
return false;
mStopReceiveMessages = false;
mThreadReceive = new fos::TaskT<SClient &>(receiveSpitMessage, *this, "receiveTask", fos::Task::DefaultStackSize, fos::Task::MAX_PRIORITY-3);
mThreadProcess = new fos::TaskT<SClient &>(processSpitMessage, *this, "processTask", fos::Task::DefaultStackSize);
return true;
}
void SClient::stopReceiveSPITMessages() {
mStopReceiveMessages = true;
}
void SClient::receiveSpitMessage(SClient &rM) {
char buffer[Message::MAX_MESSAGE_SIZE];
int rdLen;
while (rM.mStopReceiveMessages == false) {
rdLen = rM.mSocket.receiveMessage(buffer, Message::MAX_MESSAGE_SIZE, nullptr);
if (rdLen < 0) {
if (rM.mStopReceiveMessages) {
return;
}
continue;
}
auto vec = make_unique<vector<char>>(buffer, buffer + rdLen);
incommingSpit.enqueue(std::move(vec));
}
}
void SClient::processSpitMessage(SClient &rM) {
while (true) {
auto data = *incommingSpit.dequeue();
auto spitMsg = make_unique<Message>(data);
rM.checkParseMessage(move(spitMsg));
}
}
void SClient::resetMessageNumberReceived() {
mMessageNumberReceived = -1;
mProcessor.resetMessageNumberReceived();
}
void SClient::checkParseMessage(unique_ptr<Message> p_SPITMessage) {
answerWatchdog();
if (!p_SPITMessage || !p_SPITMessage->getHeader() || !p_SPITMessage->getOPPart()) {
ESP_LOGD(TAG, "Client::parseMessage incomplete message");
return;
}
int msgNum = (int)(p_SPITMessage->getHeader()->getMessageNumber() & 0xFF);
if (msgNum == 0) {
parseMessage(*p_SPITMessage);
return;
}
if (mMessageNumberReceived < 0) {
mMessageNumberReceived = msgNum;
parseMessage(*p_SPITMessage);
return;
}
int intNextMessageNumber = mMessageNumberReceived + 1;
if (intNextMessageNumber > 255 || intNextMessageNumber < 1)
intNextMessageNumber = 1;
if (msgNum == intNextMessageNumber) {
mMessageNumberReceived = msgNum;
parseMessage(*p_SPITMessage);
mMessageNumberReceived = mProcessor.sendOvertakingMessage(mMessageNumberReceived);
} else {
// p_SPITMessage has overtaken another message
mProcessor.addOvertakingMessage(move(p_SPITMessage));
}
}
void SClient::parseMessage(Message& message) {
answerWatchdog();
if (!message.getHeader() || !message.getOPPart()) {
ESP_LOGD(TAG, "Client::parseMessage incomplete message");
return;
}
switch (message.getType()) {
//----- Connection -----
case Message::ConnectAnswer::OPFLAG:
parseConnectAnswer(message.getAs<Message::ConnectAnswer>());
break;
case Message::Disconnect::OPFLAG:
parseDisconnect(message.getAs<Message::Disconnect>());
break;
case Message::Watchdog::OPFLAG:
parseWatchdog(message.getAs<Message::Watchdog>());
break;
case Message::WatchdogAnswer::OPFLAG:
parseWatchdogAnswer(message.getAs<Message::WatchdogAnswer>());
break;
default:
// pass object to SubProcessor
mProcessor.parseMessage(message);
}
}
//----- Connection ------
/**
* Tries to connect to a SPITServer.
* @param serverIP the ip address of the SPIT server
* @param serverPort the port number of the SPIT server
* @param clientIP the local IP address of the client
* @param clientPort the port number of the Client, if "0" the OS uses any possible port number
* @param loginName login user name (liveSHOW does not use this -> mayby ""
* @param loginPassword login user password (liveSHOW does not use this -> mayby ""
* @return false = an error has occured, true = the connect was sent to the server
*/
bool SClient::connect(string serverIP, string serverPort, string clientIP, string clientPort, string loginName, string loginPassword) {
disconnect();
mSocket.setConnectionParams(serverIP, serverPort, clientIP, clientPort);
if (mSocket.bindSocket(clientIP, clientPort) < 0) {
mStopReceiveMessages = true;
return false;
}
auto connMsg = mProcessor.createConnectMessage(serverIP, serverPort, loginName, loginPassword);
if (startReceiveSPITMessages() == false) {
mStopReceiveMessages = true;
return false;
}
resetMessageNumberReceived();
Connection intOldConnectionStatus = mConnectionStatus;
mConnectionStatus = Connection::TryConnection;
mCallbacks.onConnectionStatus(mServerID, mServerComputerName, mServerName, mServerStepTime,
mConnectionStatus, intOldConnectionStatus);
// sleep a little to make sure that the reception of the messages is started
fos::Task::delayMs(500);
if (sendMessage(*connMsg) < 0) {
mConnectionStatus = Connection::None;
mStopReceiveMessages = true;
return false;
}
return true;
}
/**
* Terminates the connection to the SPITServer.
*/
void SClient::disconnect() {
disconnectInternal(Connection::None);
}
void SClient::disconnectInternal(Connection cause) {
Connection intOldConnectionStatus = mConnectionStatus;
mConnectionStatus = cause;
switch (mConnectionStatus) {
case Connection::Lost:
break;
case Connection::None: {
Message::Disconnect sMDisconnect;
sendMessage(sMDisconnect);
} break;
case Connection::NoneServer:
break;
default:
break;
}
stopReceiveSPITMessages();
mCallbacks.onConnectionStatus(mServerID, mServerComputerName, mServerName, mServerStepTime,
mConnectionStatus, intOldConnectionStatus);
if (intOldConnectionStatus != Connection::None && intOldConnectionStatus != Connection::NoneServer) {
ESP_LOGI(TAG, "eventConnectionStatus %d ", (int)mConnectionStatus);
switch (mConnectionStatus) {
case SClient::Connection::None:
ESP_LOGD(TAG, "not connected");
break;
case SClient::Connection::NoneServer:
ESP_LOGD(TAG, "connection terminated by server");
// call eventProjectClosed because this may not be sended by the server,
// when the connection is closed
mCallbacks.onProjectUpdate(ProjUpdate::ServerClosed, mProcessor.getServerProjectID());
// Server has terminated the connection -> start a new game
// _doReconnect = false;
startServerAutoDetect();
break;
case SClient::Connection::Lost:
ESP_LOGD(TAG, "connection lost");
// Maybe you want to reconnect, see eventServerDetected(...)
// We have lost the connection -> set the flag to true and start the autodetection to reconnect at eventServerDetected(...)
// _doReconnect = true;
startServerAutoDetect();
break;
default:
break;
}
}
}
void SClient::parseConnectAnswer(Message::ConnectAnswer &message) {
if (message.getClientID() != mProcessor.getID())
return;
mServerID = message.getServerID();
mServerComputerName = message.getServerComputerName();
mServerName = message.getServerName();
mServerStepTime = message.getStepTime();
if (message.getConnected() == false) {
disconnectInternal(Connection::NoneServer);
} else {
mConnectionStatus = Connection::Ok;
mDisconnectedByServer = false;
ESP_LOGD(TAG, "connection established");
stopServerAutoDetect(); // normally you should stop the autodetection before you try a connection
// This is redundand
mStopReceiveMessages = false;
// Maybe you want to reconnect, see eventServerDetected(...)
// Here the connection is established -> don't reconnect at eventServerDetected(...)
// _doReconnect = false;
}
}
/**
* The SPIT server was closed
* @param message
*/
void SClient::parseDisconnect(Message::Disconnect &message) {
Connection intCauseDisconnect = Connection::NoneServer;
mDisconnectedByServer = true;
SClient::disconnectInternal(intCauseDisconnect);
}
//----- Watchdog ------
/**
* The SPIT server sends a watchdog message at regular intervals. The client responds immediately with WatchdogAnswer
* Conversely, the client also sends a watchdog message at regular intervals and waits for a WatchdogAnswer message.
* This is used to check if the other side is still present.
* @param message
*/
void SClient::parseWatchdog(Message::Watchdog &message) {
Message::WatchdogAnswer o_WatchdogAnswer;
sendMessage(o_WatchdogAnswer);
}
/**
* The SPIT server sends a watchdog message at regular intervals. The client responds immediately with WatchdogAnswer
* Conversely, the client also sends a watchdog message at regular intervals and waits for a WatchdogAnswer message.
* This is used to check if the other side is still present.
* @param message
*/
void SClient::parseWatchdogAnswer(Message::WatchdogAnswer &message) {
answerWatchdog(); // is also done by parseMessage
}
void SClient::sendWatchdog() {
int intResult;
Message::Watchdog message;
intResult = sendMessage(message);
if (intResult < 0) {
ESP_LOGW("WDT", "Failed to send Watchdog msg");
}
}
void SClient::eventWatchdogLost() {
if (mConnectionStatus == Connection::TryConnection) {
disconnectInternal(Connection::NoneServer);
} else {
disconnectInternal(Connection::Lost);
}
}
void SClient::sendProject(SClient *p_PtrSPIT_Client) {
// int intResult;
// Message::StartProjectTransfer* ptrSMStartProjectTransfer;
// Message::EndProjectTransfer* ptrSMEndProjectTransfer;
// int intCount;
// SPITType* ptrSPITType;
// SObject* ptrSPITObject;
// //StartProjectTransfer
// NotifyStartProjectTransfer_Client* o_RunnableStart = new NotifyStartProjectTransfer_Client(p_PtrSPIT_Client->getSPITClientInterface());
// SPIT_Handler::addRunnable(o_RunnableStart);
// ptrSMStartProjectTransfer = new Message::StartProjectTransfer();
// intResult = p_PtrSPIT_Client->sendMessage(ptrSMStartProjectTransfer);
// if (intResult < 0) {
// return;
// }
// // mWatchdog->answerWatchdog();
// // {
// //send all listes SPITTypes
// // unique_lock<recursive_mutex> o_LockType(*SPITType::getMutex());
// intCount = SPITType::getSPITTypeCount();
// for (int i = 0; i < intCount; i++) {
// ptrSPITType = mSpitTypes.getEntry(i);
// if (ptrSPITType == nullptr) continue;
// if (ptrSPITType->getCreatedByClient() == false) continue;
// intResult = p_PtrSPIT_Client->sendTypeInternal(ptrSPITType);
// if (intResult < 0) return;
// // mWatchdog->answerWatchdog();
// this_thread::sleep_for(chrono::milliseconds(7));
// }
// // }
// // {
// // unique_lock<recursive_mutex> o_LockObject(*SObject::getMutex());
// intCount = mSpitObjects.getEntryCount();
// for (int i = 0; i < intCount; i++) {
// ptrSPITObject = mSpitObjects.getEntry(i);
// if (ptrSPITObject == nullptr) continue;
// if (ptrSPITObject->getCreatedByClient() == false) continue;
// intResult = p_PtrSPIT_Client->sendObjectInternal(ptrSPITObject);
// if (intResult < 0) return;
// // mWatchdog->answerWatchdog();
// this_thread::sleep_for(chrono::milliseconds(7));
// }
// // }
// //EndProjectTransfer
// ptrSMEndProjectTransfer = new Message::EndProjectTransfer();
// intResult = p_PtrSPIT_Client->sendMessage(ptrSMEndProjectTransfer);
// if (intResult < 0) {
// return;
// }
// // mWatchdog->answerWatchdog();
// NotifyEndProjectTransfer_Client* o_RunnableEnd = new NotifyEndProjectTransfer_Client(p_PtrSPIT_Client->getSPITClientInterface());
// SPIT_Handler::addRunnable(o_RunnableEnd);
// //Send act
}
SType *SClient::createTypeFixed(string iD, string group, string name, uint64_t delay, uint64_t duration) {
return mProcessor.createType(iD, group, name, delay, duration, true, true);
}
fos::Semaphore SClient::sStartSad;
unordered_map<string, unique_ptr<SadDetectedServer>> SadDetectedServer::sKnownServers;
void SClient::processSocketTask(SClient &instance) {
ESP_LOGI(TAG, "Starting Socket managing");
static const string SERVERAUTODETECT_PRECODE = "#SAD#";
static const string SERVERAUTODETECT_SEPERATOR = "#,#";
static const string SERVERAUTODETECT_TASKID = "#spit#"; // -> "#SAD##,##spit#"
static const string SERVERAUTODETECT_PORT = "15110";
static constexpr int WATCHDOGTIME = 3000;
while (true) {
sStartSad.take();
bool searching = true;
instance.mSocket.setBroadcast(true);
string stringDatagram = SERVERAUTODETECT_PRECODE + SERVERAUTODETECT_SEPERATOR + SERVERAUTODETECT_TASKID;
int intDatagramLength = stringDatagram.length();
while (searching) {
if (instance.mAbortSearch)
break;
SadDetectedServer::resetServerFlags();
string stringBroadcastIP4 = instance.mSocket.mOwnIp.getBroadcastIP4();
instance.mSocket.sendMessageTo(stringBroadcastIP4, SERVERAUTODETECT_PORT, stringDatagram.data(), intDatagramLength);
{
// wait for 3000 milliseconds or
fos::Task::delayMs(3000);
}
bool receiving = true;
while (receiving) {
if (instance.mAbortSearch) {
break;
}
constexpr int SIZE = 512;
char buffer[SIZE + 1];
string stringServerIP;
int intReceivedByteCount = instance.mSocket.receiveMessage(buffer, SIZE, &stringServerIP);
if (intReceivedByteCount <= 0) {
break;
}
buffer[intReceivedByteCount] = '\0';
string stringMessage = buffer;
// Check if precode is ok
if (stringMessage.find(SERVERAUTODETECT_PRECODE.data(), 0) != 0) {
break;
}
int intIndex = 0;
int intOldPos = 0;
string stringPart;
string strParts[10];
while (intIndex < SadDetectedServer::PARAM_IP) {
int intPos = stringMessage.find(SERVERAUTODETECT_SEPERATOR.data(), intOldPos);
if (intPos == string::npos) {
stringPart = "";
} else {
stringPart = stringMessage.substr(intOldPos, intPos - intOldPos);
intOldPos = intPos + SERVERAUTODETECT_SEPERATOR.length();
}
strParts[intIndex] = stringPart;
intIndex++;
}
auto server = make_unique<SadDetectedServer>(
strParts[SadDetectedServer::PARAM_NAME],
strParts[SadDetectedServer::PARAM_ID],
strParts[SadDetectedServer::PARAM_COMPUTERNAME],
stringServerIP,
strParts[SadDetectedServer::PARAM_PORT],
strParts[SadDetectedServer::PARAM_CLIENTIP],
strParts[SadDetectedServer::PARAM_ADDITIONS],
strParts[SadDetectedServer::PARAM_SERVERPROJECTNAME]);
SadDetectedServer::addDetectedServer(move(server));
}
fos::Task::delayMs(100);
SadDetectedServer::evaluateDetectedServers(instance);
instance.mAbortSearch = false;
// if(instance.mConnectionStatus == Connection::Ok) break;
fos::Task::delayMs(1000);
}
instance.mSocket.setBroadcast(false);
ESP_LOGW(TAG, "Finished searching");
ESP_LOGW(TAG, "Start working");
auto nextWdTrigger = fos::Timing::getMilliSec() + WATCHDOGTIME;
// auto nextWdCheck = nextWdTrigger + WATCHDOGTIME;
while (true) {
// auto time = fos::Timing::getMilliSec();
// if (time > nextWdTrigger) {
// instance.sAnswered = false;
// // instance.sendWatchdog();
// nextWdTrigger = nextWdCheck + WATCHDOGTIME;
// }
// if (time > nextWdCheck) {
// if (!instance.sAnswered) {
// ESP_LOGW(TAG, "TimeOut");
// instance.eventWatchdogLost();
// break;
// }
// nextWdCheck = nextWdTrigger + WATCHDOGTIME;
// }
fos::Task::delayMs(25);
// if (instance.mStopReceiveMessages)
// break;
}
}
}
}