diff --git a/components/constants/include/constants/tcp.h b/components/constants/include/constants/tcp.h index bd4ec8f..a061786 100644 --- a/components/constants/include/constants/tcp.h +++ b/components/constants/include/constants/tcp.h @@ -13,6 +13,8 @@ #define SLEEP_AFTER_FAIL_MS 5000 +#define NO_CLIENT_SLEEP_MS 400 + #define TCP_PORT 3001 #endif //TCP_H diff --git a/components/rpc/CommunicationRouter.cpp b/components/rpc/CommunicationRouter.cpp index 7d60024..89e1d95 100644 --- a/components/rpc/CommunicationRouter.cpp +++ b/components/rpc/CommunicationRouter.cpp @@ -84,8 +84,12 @@ void CommunicationRouter::update_leader() { if (this->m_leader != max) { if (max == m_module_id) { m_pc_connection->connect(); + m_lossless_server->startup(); + // todo: BTS-22 add lossy server } else if (this->m_leader == m_module_id) { m_pc_connection->disconnect(); + m_lossless_server->shutdown(); + // todo: BTS-22 add lossy server } } diff --git a/components/rpc/MessagingInterface.cpp b/components/rpc/MessagingInterface.cpp index 99bef0e..97b11b9 100644 --- a/components/rpc/MessagingInterface.cpp +++ b/components/rpc/MessagingInterface.cpp @@ -2,6 +2,8 @@ // Created by Johnathon Slightham on 2025-05-25. // +#include + #include "MessagingInterface.h" #include "AngleControlMessageBuilder.h" #include "ConfigManager.h" diff --git a/components/rpc/include/IConnectionManager.h b/components/rpc/include/IConnectionManager.h index 1065612..de359d4 100644 --- a/components/rpc/include/IConnectionManager.h +++ b/components/rpc/include/IConnectionManager.h @@ -12,4 +12,4 @@ public: virtual int disconnect() = 0; }; -#endif //IWIFIMANAGER_H +#endif //ICONNECTIONMANAGER_H diff --git a/components/rpc/include/IRPCServer.h b/components/rpc/include/IRPCServer.h index ed91386..5efe3b0 100644 --- a/components/rpc/include/IRPCServer.h +++ b/components/rpc/include/IRPCServer.h @@ -8,6 +8,8 @@ class IRPCServer { public: virtual ~IRPCServer() = default; + virtual void startup() = 0; + virtual void shutdown() = 0; virtual int send_msg(char* buffer, uint32_t length) const = 0; }; diff --git a/components/rpc/include/wireless/TCPServer.h b/components/rpc/include/wireless/TCPServer.h index 299fb81..cd63aee 100644 --- a/components/rpc/include/wireless/TCPServer.h +++ b/components/rpc/include/wireless/TCPServer.h @@ -17,6 +17,8 @@ class TCPServer final : public IRPCServer { public: TCPServer(int port, const std::shared_ptr>>& rx_queue); ~TCPServer() override; + void startup() override; + void shutdown() override; int send_msg(char* buffer, uint32_t length) const override; private: diff --git a/components/rpc/wireless/TCPServer.cpp b/components/rpc/wireless/TCPServer.cpp index 2577b6a..85ca928 100644 --- a/components/rpc/wireless/TCPServer.cpp +++ b/components/rpc/wireless/TCPServer.cpp @@ -1,4 +1,3 @@ -#include #include #include "esp_log.h" @@ -18,6 +17,8 @@ #define TAG "TCPServer" +#define MIN(X, Y) (((X) < (Y)) ? (X) : (Y)) + // todo: - add message routing to correct client // - authenticate (don't just return true from the auth function) // - tx from board @@ -30,15 +31,37 @@ TCPServer::TCPServer(const int port, const std::shared_ptrm_rx_task = nullptr; this->m_rx_queue = rx_queue; this->m_server_sock = 0; +} + +TCPServer::~TCPServer() { + this->shutdown(); + vSemaphoreDelete(this->m_mutex); +} + +void TCPServer::startup() { + ESP_LOGI(TAG, "Starting TCP server on port %d", this->m_port); + if (nullptr != this->m_task || nullptr != this->m_rx_task) { + ESP_LOGW(TAG, "Attempted to start TCP server when already started, ignoring start request"); + return; + } xTaskCreate(tcp_server_task, "tcp_accept_server", 3072, this, 5, &this->m_task); xTaskCreate(socket_monitor_thread, "tcp_rx", 4096, this, 5, &this->m_rx_task); } -TCPServer::~TCPServer() { - vTaskDelete(this->m_task); - vTaskDelete(this->m_rx_task); - vSemaphoreDelete(this->m_mutex); +void TCPServer::shutdown() { + ESP_LOGI(TAG, "Shutting down TCP server"); + if (nullptr != this->m_task) { + vTaskDelete(this->m_task); + close(this->m_server_sock); + } + + if (nullptr != this->m_rx_task) { + vTaskDelete(this->m_rx_task); + for (const auto sock : this->m_clients) { + close(sock); + } + } } [[noreturn]] void TCPServer::tcp_server_task(void *args) { @@ -131,15 +154,16 @@ TCPServer::~TCPServer() { int max_fd = -1; xSemaphoreTake(that->m_mutex, portMAX_DELAY); + if (that->m_clients.size() < 1) { + vTaskDelay(NO_CLIENT_SLEEP_MS / portTICK_PERIOD_MS); + } for (const auto sock : that->m_clients) { FD_SET(sock, &readfds); if (sock > max_fd) max_fd = sock; } xSemaphoreGive(that->m_mutex); - // todo: Select seems to be timing out the watchdog, even though we have a 50ms timeout. - // Potentially select is not respecting our timeout? - timeval timeout = {0, 50000}; // 50 ms timeout + timeval timeout = {.tv_sec = 1, .tv_usec = 0}; // 1s timeout int ret = select(max_fd + 1, &readfds, nullptr, nullptr, &timeout); vTaskDelay(0); // Avoid starving other threads @@ -150,17 +174,25 @@ TCPServer::~TCPServer() { for (int sock : that->m_clients) { vTaskDelay(0); // Avoid starving other threads if (FD_ISSET(sock, &readfds)) { - // Handle socket - auto buffer = std::make_unique>(); - buffer->resize(MAX_RX_BUFFER_SIZE); uint32_t msg_size = 0; - recv(sock, &msg_size, 4, MSG_WAITALL); - if (msg_size < 1 || msg_size > 512) { + if (int len = recv(sock, &msg_size, 4, MSG_WAITALL); len < 0) { + ESP_LOGE(TAG, "Error occurred during receiving msg length: errno %d\n", errno); + to_remove.emplace_back(sock); + continue; + } else if (0 == len) { + ESP_LOGI(TAG, "TCP Connection closed when receiving msg length\n"); + close(sock); + to_remove.emplace_back(sock); continue; } - ESP_LOGD(TAG, "Message size: %ld\n", msg_size); + if (msg_size < 1 || msg_size > MAX_RX_BUFFER_SIZE) { + continue; + } + + auto buffer = std::make_unique>(); + buffer->resize(MIN(MAX_RX_BUFFER_SIZE, msg_size)); if (int len = recv(sock, buffer->data(), msg_size, MSG_WAITALL); len < 0) { ESP_LOGE(TAG, "Error occurred during receiving: errno %d\n", errno); @@ -179,6 +211,7 @@ TCPServer::~TCPServer() { for (const auto r : to_remove) { that->m_clients.erase(r); + close(r); } xSemaphoreGive(that->m_mutex); @@ -228,4 +261,3 @@ int TCPServer::send_msg(char *buffer, const uint32_t length) const { return 0; } -