MPI

components/rpc/include/CommunicationRouter.h

@@ -5,12 +5,34 @@
 #ifndef COMMUNICATIONROUTER_H
 #define COMMUNICATIONROUTER_H
 
+#include <functional>
+#include <memory>
 
+#include "TCPServer.h"
+#include "constants/tcp.h"
+#include "constants/app_comms.h"
 
-class CommunicationRouter : ICommunicationRouter {
+class CommunicationRouter {
+public:
+    explicit CommunicationRouter(const std::function<void(char*, int)> &rx_callback)
+        : m_tcp_rx_queue(xQueueCreate(RX_QUEUE_SIZE, MAX_RX_BUFFER_SIZE)),
+            m_rx_callback(rx_callback),
+            m_tcp_server(std::make_unique<TCPServer>(TCP_PORT, m_tcp_rx_queue)) {
+        xTaskCreate(router_thread, "communication_router", 2048, this, 3, &this->m_router_thread);
+    }
+
+    ~CommunicationRouter();
+
+    [[noreturn]] static void router_thread(void *args);
+
+    // todo: does this really need to be here (so i can access from thread)?
+    QueueHandle_t m_tcp_rx_queue;
+    std::function<void(char*, int)> m_rx_callback;
+private:
+
+    TaskHandle_t m_router_thread;
+    std::unique_ptr<TCPServer> m_tcp_server;
 
 };
 
-
-
 #endif //COMMUNICATIONROUTER_H

components/rpc/include/ICommunicationRouter.h

@@ -1,12 +0,0 @@
-//
-// Created by Johnathon Slightham on 2025-05-25.
-//
-
-#ifndef ICOMMUNICATIONROUTER_H
-#define ICOMMUNICATIONROUTER_H
-
-class ICommunicationRouter {
-
-}
-
-#endif //ICOMMUNICATIONROUTER_H

components/rpc/include/MessagingInterface.h

@@ -5,12 +5,35 @@
 #ifndef MESSAGINGINTERFACE_H
 #define MESSAGINGINTERFACE_H
 
+#include <memory>
+#include <unordered_map>
 
+#include "CommunicationRouter.h"
 
 class MessagingInterface {
+public:
+    MessagingInterface()
+        : m_mpi_rx_queue(xQueueCreate(MAX_RX_BUFFER_SIZE, RX_QUEUE_SIZE)),
+            m_router(std::make_unique<CommunicationRouter>([this](char* buffer, const int size) { handleRecv(buffer, size); })),
+            m_map_semaphore(xSemaphoreCreateMutex()) {};
 
+    ~MessagingInterface();
+
+    int send(char* buffer, int size, int destination, int tag, bool durable);
+    int broadcast(char* buffer, int size, int root, bool durable);
+    int recv(char* buffer, int size, int source, int tag);
+    int sendrecv(char* send_buffer, int send_size, int dest, int send_tag, char* recv_buffer, int recv_size, int recv_tag);
+
+private:
+    void handleRecv(const char* recv_buffer, int recv_size);
+
+    void checkOrInsertTag(uint8_t tag);
+
+    uint16_t sequence_number = 0;
+    QueueHandle_t m_mpi_rx_queue; // todo: maybe move this down classes more
+    std::unique_ptr<CommunicationRouter> m_router;
+    SemaphoreHandle_t m_map_semaphore;
+    std::unordered_map<uint8_t, QueueHandle_t> m_tag_to_queue;
 };
 
-
-
 #endif //MESSAGINGINTERFACE_H

components/rpc/include/TCPServer.h

@@ -14,7 +14,7 @@
 
 class TCPServer : IRPCServer {
 public:
-    explicit TCPServer(int port);
+    TCPServer(int port, QueueHandle_t rx_queue);
     ~TCPServer();
 
 private:
@@ -31,6 +31,8 @@ private:
     TaskHandle_t m_rx_task;
     TaskHandle_t m_tx_task;
 
+    QueueHandle_t m_rx_queue;
+
     SemaphoreHandle_t m_mutex;
     std::unordered_set<int> m_clients;
 };

components/rpc/include/rpc.h

@@ -1 +0,0 @@
-void func(void);