abstract common tcp operations to a base class

client/src/TcpClient.cpp

@@ -1,7 +1,7 @@
 
 #include "TcpClient.hpp"
 
-TcpClient::TcpClient(ConfigService* config, LoggerService* logger) : config_(config), logger_(logger) {
+TcpClient::TcpClient(ConfigService* config, LoggerService* logger) : TcpBase(logger) {
 
     if(!(config->getConfig<ClientConfig>("TcpClient", "main", &configuration_))) {
         logger_->log("TcpClient", LogFlag::Error, "Failed to get configuration");
@@ -61,9 +61,14 @@ ErrorCode TcpClient::init() {
         return ErrorCode::Error;
     }
     
+    std::string message = "placeholder";
+    while(!message.empty()) {
+        logger_->log("TcpClient", LogFlag::Debug, "Enter message...");
+        std::getline(std::cin, message);
+
         // send data buffer to server
-    memset(buffer, 0x67, sizeof(buffer));
+        sendBytes = tcpSend(socketFd, message.c_str(), message.size());
-    sendBytes = send(socketFd, buffer, sizeof(buffer), 0);
+
         if(sendBytes < 0) {
             logger_->log("TcpClient", LogFlag::Error, "Unable to send to server.");
             close(socketFd);
@@ -74,7 +79,7 @@ ErrorCode TcpClient::init() {
 
         // listen for receive from the server
         memset(buffer, 0x00, sizeof(buffer));
-    receiveBytes = recv(socketFd, buffer, sizeof(buffer), 0); // TODO: this currently blocks forever, add timeout
+        receiveBytes = tcpRead(socketFd, buffer, sizeof(buffer)); // TODO: this currently blocks forever, add timeout
 
         // we expect a response based on the server design, something went wrong otherwise
         if(receiveBytes < 0) {
@@ -87,6 +92,7 @@ ErrorCode TcpClient::init() {
         }
 
         logger_->log("TcpClient", LogFlag::Debug, "Received {} bytes back from the server", receiveBytes);
+    }
 
     close(socketFd);
 

client/src/TcpClient.hpp

@@ -11,9 +11,10 @@
 #include "common/config/ConfigService.hpp"
 #include "common/LoggerService.hpp"
 #include "common/ErrorCodes.hpp"
+#include "common/TcpBase.hpp"
 #include "config/ClientConfig.hpp"
 
-class TcpClient {
+class TcpClient : TcpBase {
 
 public:
 
@@ -24,8 +25,6 @@ private:
 
     ErrorCode init();
 
-    LoggerService* logger_;
-    ConfigService* config_;
     ClientParams configuration_;
 
 };

common/CMakeLists.txt

@@ -1,6 +1,7 @@
 
 add_library(accordion-common STATIC
     LoggerService.cpp
+    TcpBase.cpp
     config/ConfigService.cpp
 )
 

common/LoggerService.cpp

@@ -101,5 +101,7 @@ void LoggerService::write(std::string component, LogFlag flag, std::string messa
     if(configuration_.fileEnabled) {
         outfile_ << finalmessage << std::endl;
     }
+
+    // TODO: if its an error flag we should automatically log the errno
 }
 

common/TcpBase.cpp

@@ -0,0 +1,121 @@
+
+#include "TcpBase.hpp"
+
+#include <sys/socket.h>
+#include <string.h>
+
+TcpBase::TcpBase(LoggerService* logger) : logger_(logger) {
+
+}
+
+TcpBase::~TcpBase() {
+
+}
+
+ssize_t TcpBase::tcpRead(int fd, void* buffer, size_t maxBytes) {
+
+    // read header first
+    char header[kHeaderSize];
+    ssize_t bytesReceived = receiveFull(fd, &header, kHeaderSize);
+    if(bytesReceived != kHeaderSize) {
+        logger_->log("TcpNetworker", LogFlag::Error, "Unable to receive header.");
+        return -1;
+    }
+
+    // parse header: 4 bytes payload length 4 bytes message type
+    uint32_t payloadLength;
+    uint32_t messageType;
+    memcpy(&payloadLength, header, sizeof(payloadLength));
+    memcpy(&messageType, header + sizeof(payloadLength), sizeof(messageType));
+    size_t payloadLengthS = static_cast<size_t>(payloadLength);
+
+    if(maxBytes < payloadLength) {
+        logger_->log("TcpNetworker", LogFlag::Warning,
+            "Payload length is larger than buffer size; payload may be truncated.");
+        payloadLengthS = maxBytes;
+    }
+
+    // read payload
+    bytesReceived = receiveFull(fd, buffer, payloadLengthS);
+    if(bytesReceived != payloadLengthS) {
+        logger_->log("TcpNetworker", LogFlag::Error, "Did not receive amount of expected data.");
+    }
+    return bytesReceived;
+
+}
+
+ssize_t TcpBase::tcpTimedRead(int fd, void* buffer, size_t maxBytes, int timeoutPlaceholder) {
+    return 0;
+}
+
+ssize_t TcpBase::tcpSend(int fd, const void* buffer, size_t numBytes) {
+
+    // assemble header
+    char header[kHeaderSize];
+    uint32_t payloadLength = static_cast<uint32_t>(numBytes & 0xffffffffffffffff);
+    uint32_t messageType = 0; // placeholder
+    memcpy(header, &payloadLength, sizeof(payloadLength));
+    memcpy(header + sizeof(payloadLength), &messageType, sizeof(messageType));
+
+    // send header
+    ssize_t bytesSent = sendFull(fd, header, kHeaderSize);
+    if(bytesSent != kHeaderSize) {
+        logger_->log("TcpNetworker", LogFlag::Error, "Error sending header.");
+    }
+
+    // send payload
+    bytesSent = sendFull(fd, buffer, numBytes);
+    if(bytesSent != numBytes) {
+        logger_->log("TcpNetworker", LogFlag::Error, "Error sending payload.");
+    }
+    return bytesSent;
+}
+
+ssize_t TcpBase::tcpTimedSend(int fd, const void* backuffer, size_t numBytes, int timeoutPlaceholder) {
+    return 0;
+}
+
+ssize_t TcpBase::receiveHeader(int fd, void* buffer, size_t maxBytes) {
+    return 0;    
+}
+
+ssize_t TcpBase::receivePayload(int fd, void* buffer, size_t maxBytes) {
+    return 0;
+}
+
+ssize_t TcpBase::sendHeader(int fd, const void* buffer, size_t maxBytes) {
+    return 0;
+}
+
+ssize_t TcpBase::sendPayload(int fd, const void* buffer, size_t maxBytes) {
+    return 0;
+}
+
+ssize_t TcpBase::receiveFull(int fd, void* buffer, size_t maxBytes) {
+    ssize_t total = 0;
+    // attempt to receive maxBytes bytes; continue receiving until we have received maxBytes bytes
+    while(total < maxBytes) {
+        ssize_t bytesReceived = recv(fd, static_cast<char*>(buffer) + total, maxBytes - total, 0);
+        if (bytesReceived <= 0) return bytesReceived; // error or connection closed
+        total += bytesReceived;
+    }
+
+    return total;
+}
+
+ssize_t TcpBase::sendFull(int fd, const void* buffer, size_t numBytes) {
+    ssize_t total = 0;
+    // attempt to receive maxBytes bytes; continue receiving until we have received maxBytes bytes
+    while(total < numBytes) {
+        ssize_t bytesSent = send(fd, static_cast<const char*>(buffer) + total, numBytes - total, 0);
+
+        // last arg is flags, 0=default
+        // MSG_PEEK: read without consuming the message from the queue
+        // MSG_WAITALL: block until all of the specified size is received
+        // MSG_DONTWAIT: non-blocking receive, returns -1 immediately if queue is empty
+
+        if (bytesSent <= 0) return bytesSent; // error or connection closed
+        total += bytesSent;
+    }
+    return total;
+}

common/TcpBase.hpp

@@ -0,0 +1,44 @@
+
+#pragma once
+
+#include "ErrorCodes.hpp"
+#include "LoggerService.hpp"
+
+// TcpBase contains common tcp I/O operations for the tcp client and tcp server
+// Abstracts away the networking methods into an accordion common protocol (ACP)
+class TcpBase {
+
+public:
+
+    TcpBase(LoggerService* logger);
+    ~TcpBase();
+
+    // read from a tcp socket, blocking
+    ssize_t tcpRead(int fd, void* buffer, size_t maxBytes);
+
+    // read from a tcp socket, blocking with timeout
+    ssize_t tcpTimedRead(int fd, void* buffer, size_t maxBytes, int timeoutPlaceholder);
+
+    // send over a tcp socket, blocking
+    ssize_t tcpSend(int fd, const void* buffer, size_t numBytes);
+
+    // send over a tcp socket, blocking with timeout
+    ssize_t tcpTimedSend(int fd, const void* buffer, size_t numBytes, int timeoutPlaceholder);
+
+protected:
+
+    LoggerService* logger_;
+
+private:
+
+    ssize_t receiveHeader(int fd, void* buffer, size_t maxBytes);
+    ssize_t receivePayload(int fd, void* buffer, size_t maxBytes);
+    ssize_t sendHeader(int fd, const void* buffer, size_t maxBytes);
+    ssize_t sendPayload(int fd, const void* buffer, size_t maxBytes);
+
+    ssize_t receiveFull(int fd, void* buffer, size_t maxBytes);
+    ssize_t sendFull(int fd, const void* buffer, size_t numBytes);
+
+    static constexpr size_t kHeaderSize = 8; // 4 bytes for length, 4 bytes for type
+
+};

server/src/TcpServer.cpp

@@ -1,7 +1,7 @@
 
 #include "TcpServer.hpp"
 
-TcpServer::TcpServer(ConfigService* config, LoggerService* logger) : config_(config), logger_(logger) {
+TcpServer::TcpServer(ConfigService* config, LoggerService* logger) : TcpBase(logger) {
 
     if(!(config->getConfig<ServerConfig>("TcpServer", "main", &configuration_))) {
         logger_->log("TcpServer", LogFlag::Error, "Failed to get configuration");
@@ -79,6 +79,7 @@ ErrorCode TcpServer::init() {
             // TODO: more robust errorchecking with errnos
             return ErrorCode::Error;
         }
+        // TODO: because accept blocks, the server can't handle parallel clients.
         logger_->log("TcpServer", LogFlag::Info
             , "Client connected from {}:{}",
             inet_ntoa(clientAddress.sin_addr), ntohs(clientAddress.sin_port));
@@ -86,17 +87,19 @@ ErrorCode TcpServer::init() {
         // receive loop
         while(1) { 
             // receive messages from the client. blocks until there's a message to receive
-            receiveBytes = recv(clientSocket, buffer, sizeof(buffer), 0);
+            receiveBytes = tcpRead(clientSocket, buffer, sizeof(buffer));
-            // last arg is flags, 0=default
+            std::string message(buffer, receiveBytes);
-            // MSG_PEEK: read without consuming the message from the queue
-            // MSG_WAITALL: block until all of the specified size is received
-            // MSG_DONTWAIT: non-blocking receive, returns -1 immediately if queue is empty
-            if(receiveBytes <= 0) break; // exit when there's nothing left to receive
 
-            logger_->log("TcpClient", LogFlag::Debug, "Received {} bytes from client: 0x{:x}", receiveBytes, buffer[0]);
+            // everything below is a simulated "processMessage()"
+            logger_->log("TcpClient", LogFlag::Debug, "Received {} bytes from client {}:{}: {}",
+                receiveBytes, inet_ntoa(clientAddress.sin_addr), ntohs(clientAddress.sin_port), message);
 
             // echo back received data back to the client
-            sendBytes = send(clientSocket, buffer, receiveBytes, 0); // TODO: send size is not guarenteed
+            sendBytes = tcpSend(clientSocket, buffer, receiveBytes);
+            // TODO: send size is not guarenteed
+            // the solution might be a standard header that contains message type, checksum, and message length
+            // i.e. first four bytes are the total length of the message
+
             if(sendBytes < 0) {
                 logger_->log("TcpServer", LogFlag::Error, "Unable to send to client.");
                 return ErrorCode::Error;

server/src/TcpServer.hpp

@@ -12,9 +12,10 @@
 #include "common/config/ConfigService.hpp"
 #include "common/LoggerService.hpp"
 #include "common/ErrorCodes.hpp"
+#include "common/TcpBase.hpp"
 #include "config/ServerConfig.hpp"
 
-class TcpServer {
+class TcpServer : TcpBase {
 
 public:
 
@@ -25,8 +26,6 @@ private:
 
     ErrorCode init();
 
-    LoggerService* logger_;
-    ConfigService* config_;
     ServerParams configuration_;
 
 };