diff --git a/flake.lock b/flake.lock index 246cfd4..8abbdfa 100644 --- a/flake.lock +++ b/flake.lock @@ -16,9 +16,44 @@ "type": "github" } }, + "nixpkgs_2": { + "locked": { + "lastModified": 1784356753, + "narHash": "sha256-12KrbMiWLcf8m7pCvAtZh1ZrgF85ZXDXvfR/fWTKy84=", + "owner": "nixos", + "repo": "nixpkgs", + "rev": "61b7c44c4073f0b827768aff0049561b5110ea5a", + "type": "github" + }, + "original": { + "owner": "nixos", + "ref": "nixos-unstable", + "repo": "nixpkgs", + "type": "github" + } + }, "root": { "inputs": { - "nixpkgs": "nixpkgs" + "nixpkgs": "nixpkgs", + "tablog": "tablog" + } + }, + "tablog": { + "inputs": { + "nixpkgs": "nixpkgs_2" + }, + "locked": { + "lastModified": 1784813121, + "narHash": "sha256-ixwsmElsnlQE0MHPjS6m5+XmS47bFPHBYtZIxCTwLB8=", + "owner": "Sobottasgithub", + "repo": "tablog", + "rev": "e47b0796e6a0820315a846fb02a9c9c89b5bda33", + "type": "github" + }, + "original": { + "owner": "Sobottasgithub", + "repo": "tablog", + "type": "github" } } }, diff --git a/flake.nix b/flake.nix index 6b47b5d..ca39332 100644 --- a/flake.nix +++ b/flake.nix @@ -3,15 +3,25 @@ inputs = { nixpkgs.url = "github:nixos/nixpkgs?ref=nixos-unstable"; + + tablog = { + url = "github:Sobottasgithub/tablog"; + }; }; outputs = - { self, nixpkgs }: + { + self, + nixpkgs, + tablog, + }: let system = "x86_64-linux"; pkgs = import nixpkgs { inherit system; }; - version = "2.0.1"; + version = "2.1.7"; + + libtablog = tablog.packages.${system}.lib; commonDeps = with pkgs; [ cmake @@ -19,6 +29,7 @@ gnumake libtasn1 arrow-cpp + libtablog ]; mkTTP2Package = @@ -67,7 +78,7 @@ }; in { - inherit lib; + inherit lib libtablog; client = mkTTP2Package { pname = "ttp2-client"; diff --git a/lib/ttp2/CMakeLists.txt b/lib/ttp2/CMakeLists.txt index e2449f7..d83c6cf 100644 --- a/lib/ttp2/CMakeLists.txt +++ b/lib/ttp2/CMakeLists.txt @@ -24,6 +24,7 @@ target_include_directories(ttp2 PUBLIC # Link against the libtasn1 library target_link_libraries(ttp2 PRIVATE tasn1) +target_link_libraries(ttp2 PRIVATE tablog) install(TARGETS ttp2 DESTINATION lib) install(DIRECTORY include/ DESTINATION include) diff --git a/lib/ttp2/include/asn1_helpers.h b/lib/ttp2/include/asn1_helpers.h index 7f4a3c8..d378e1f 100644 --- a/lib/ttp2/include/asn1_helpers.h +++ b/lib/ttp2/include/asn1_helpers.h @@ -1,6 +1,8 @@ #ifndef ASN1_HELPERS_H #define ASN1_HELPERS_H +#include + #include #include #include diff --git a/lib/ttp2/include/client_session_controller.h b/lib/ttp2/include/client_session_controller.h index 1575e9e..7636384 100644 --- a/lib/ttp2/include/client_session_controller.h +++ b/lib/ttp2/include/client_session_controller.h @@ -11,6 +11,7 @@ namespace ttp2 { ClientSessionController(); ClientSessionController(int &socket); void networkingSession(); + void disconnect() override; private: int socket; diff --git a/lib/ttp2/include/networking.h b/lib/ttp2/include/networking.h index d552e5d..94c2756 100644 --- a/lib/ttp2/include/networking.h +++ b/lib/ttp2/include/networking.h @@ -1,6 +1,8 @@ #ifndef NETWORKING_H #define NETWORKING_H +#include + #include #include #include @@ -14,7 +16,8 @@ namespace ttp2 { class Networking { - public: + public: + // Packet Payloads struct Standard { std::string payload = ""; }; @@ -23,18 +26,30 @@ namespace ttp2 { std::string filePath = ""; int start = -1; int end = -1; - std::shared_ptr payload; + std::shared_ptr payload = arrow::Table::Make(arrow::schema({}), std::vector>{}, 0); }; + struct ViewportRequest { + int xStart = 0; + int xEnd = 0; + int yStart = 0; + int yEnd = 0; + }; + struct Viewport { int xStart = 0; int xEnd = 0; int yStart = 0; int yEnd = 0; - std::shared_ptr payload; + std::shared_ptr payload = arrow::Table::Make(arrow::schema({}), std::vector>{}, 0); + }; + + struct Filter { + std::string columnName; + std::string regex; }; - typedef std::variant payloadVariants; + typedef std::variant payloadVariants; struct Packet { int id = -1; @@ -42,7 +57,6 @@ namespace ttp2 { }; bool isConnected(); - void disconnect(); bool hasRequest(); bool hasResponse(); Packet popRequest(); @@ -56,19 +70,35 @@ namespace ttp2 { int sendPacket(int socket, Packet packet); Packet receiveMessage(int socket); + // WARNING: This struct cant be send as a payload type! + struct PacketInfo { + int id; + payloadVariants payloadType; + }; + PacketInfo peekResponse(); + PacketInfo peekResponse(int index); + PacketInfo peekRequest(); + PacketInfo peekRequest(int index); + static std::string getBroadcastIpAddress(); static std::string getLocalIpAddress(std::string interface); static bool isValidIpV4(std::string &ipString); static bool isValidInterface(std::string &interface); + + virtual void disconnect(); + + protected: + std::shared_ptr logger; - protected: bool connected = true; bool isNumeric(const std::string& string); int bytesToInt(std::vector bytes, int size); - static std::shared_ptr tableToBuffer(const std::shared_ptr& table); - static std::shared_ptr bufferToTable(const uint8_t* rawData, int64_t dataSize); + std::shared_ptr tableToBuffer(const std::shared_ptr& table); + std::shared_ptr bufferToTable(const uint8_t* rawData, int64_t dataSize); + + void configureLogger(std::string name); std::vector requestQueue; std::vector responseQueue; @@ -77,7 +107,7 @@ namespace ttp2 { std::map> sessionBuffers; - int autoId; + int autoId = 0; }; } diff --git a/lib/ttp2/include/server_session_controller.h b/lib/ttp2/include/server_session_controller.h index 7ff82b9..4266847 100644 --- a/lib/ttp2/include/server_session_controller.h +++ b/lib/ttp2/include/server_session_controller.h @@ -11,6 +11,8 @@ namespace ttp2 { ServerSessionController(); ServerSessionController(int serverSocket, int clientSocket); void networkingSession(); + void disconnect() override; + private: int serverSocket; int clientSocket; diff --git a/lib/ttp2/src/asn1_helpers.cpp b/lib/ttp2/src/asn1_helpers.cpp index 1bb4243..6d8d59b 100644 --- a/lib/ttp2/src/asn1_helpers.cpp +++ b/lib/ttp2/src/asn1_helpers.cpp @@ -1,5 +1,8 @@ #include "../include/asn1_helpers.h" +#include +#include + #include #include #include @@ -21,18 +24,21 @@ namespace ttp2 { standardPayload, std::strlen(standardPayload)); if (status != ASN1_SUCCESS) { - std::wcout << "ASN1 set " << asn1Key << " failed!" << std::endl; + std::string asn1KeyString = asn1Key; + tablog::TablogRegistry::getInstance().get("TTP2")->log(tablog::ERROR, "ASN1 set " + asn1KeyString + " failed!"); } return packet; } asn1_node Asn1Helpers::asn1EncodePayload(int payload, asn1_node packet, const char* asn1Key) { + std::string payloadString = std::to_string(payload); int status = asn1_write_value(packet, asn1Key, - &payload, sizeof(payload)); + payloadString.c_str(), 0); if (status != ASN1_SUCCESS) { - std::wcout << "ASN1 set " << asn1Key << " failed!" << std::endl; + std::string asn1KeyString = asn1Key; + tablog::TablogRegistry::getInstance().get("TTP2")->log(tablog::ERROR, "ASN1 set " + asn1KeyString + " failed!"); } return packet; @@ -44,7 +50,8 @@ namespace ttp2 { int status = asn1_write_value(packet, asn1Key, targetBuffer, size); if (status != ASN1_SUCCESS) { - std::wcout << "ASN1 set " << asn1Key << " failed!" << std::endl; + std::string asn1KeyString = asn1Key; + tablog::TablogRegistry::getInstance().get("TTP2")->log(tablog::ERROR, "ASN1 set " + asn1KeyString + " failed!"); } return packet; @@ -89,11 +96,11 @@ namespace ttp2 { std::vector Asn1Helpers::asn1DecodePayloadBuffer(asn1_node packet, const char* asn1Key) { int payloadLen = 0; - int status2 = asn1_read_value(packet, asn1Key, nullptr, &payloadLen); + int status = asn1_read_value(packet, asn1Key, nullptr, &payloadLen); std::vector buffer(payloadLen); - if (status2 == ASN1_MEM_ERROR && payloadLen > 0) { - status2 = asn1_read_value(packet, asn1Key, buffer.data(), &payloadLen); + if (status == ASN1_MEM_ERROR && payloadLen > 0) { + status = asn1_read_value(packet, asn1Key, buffer.data(), &payloadLen); return buffer; } return buffer; diff --git a/lib/ttp2/src/client_session_controller.cpp b/lib/ttp2/src/client_session_controller.cpp index 485b1d8..854db9e 100644 --- a/lib/ttp2/src/client_session_controller.cpp +++ b/lib/ttp2/src/client_session_controller.cpp @@ -1,5 +1,7 @@ #include "../include/client_session_controller.h" +#include + #include #include #include @@ -8,22 +10,25 @@ #include namespace ttp2 { - ClientSessionController::ClientSessionController() {} + ClientSessionController::ClientSessionController() { + configureLogger("TTP2"); + } ClientSessionController::ClientSessionController(int &socket) { + configureLogger("TTP2"); this->socket = socket; } void ClientSessionController::networkingSession() { epollFd = epoll_create1(0); if (epollFd == -1) { - std::wcout << "Failed to create epoll!" << std::endl; + logger->log(tablog::ERROR, "Failed to create epoll!"); } serverEvent.events = EPOLLIN; serverEvent.data.fd = socket; if (epoll_ctl(epollFd, EPOLL_CTL_ADD, socket, &serverEvent) == -1) { - std::wcout << "Failed to set epoll_ctl for client!" << std::endl; + logger->log(tablog::ERROR, "Failed to set epoll_ctl for client!"); return; } @@ -49,10 +54,10 @@ namespace ttp2 { } void ClientSessionController::receiveResponseSession() { + const int MAX_EVENTS = 10; + while (isConnected()) { - const int MAX_EVENTS = 10; struct epoll_event incomingEvents[MAX_EVENTS]; - int eventCount = epoll_wait(epollFd, incomingEvents, MAX_EVENTS, -1); for (int index = 0; index < eventCount; ++index) { @@ -63,13 +68,11 @@ namespace ttp2 { continue; } if (incomingEvents[index].events & EPOLLIN) { - while (true) { - Packet packet = receiveMessage(fd); - if (packet.id == -1) { - break; - } - pushResponse(packet); + Packet packet = receiveMessage(fd); + if (packet.id == -1) { + continue; } + pushResponse(packet); } } } @@ -83,4 +86,10 @@ namespace ttp2 { } } } + + void ClientSessionController::disconnect() { + std::lock_guard lock(mtx); + close(this->socket); + connected = false; + } } diff --git a/lib/ttp2/src/networking.cpp b/lib/ttp2/src/networking.cpp index a5514a9..74a9504 100644 --- a/lib/ttp2/src/networking.cpp +++ b/lib/ttp2/src/networking.cpp @@ -1,14 +1,10 @@ #include "../include/networking.h" #include "../include/asn1_helpers.h" +#include +#include + #include -#include -#include -#include -#include -#include -#include -#include #include #include #include @@ -19,6 +15,7 @@ #include #include #include +#include #include #include #include @@ -26,6 +23,14 @@ #include #include #include +#include +#include +#include +#include +#include +#include +#include + extern "C" { #include @@ -48,11 +53,6 @@ namespace ttp2 { return connected; } - void Networking::disconnect() { - std::lock_guard lock(mtx); - connected = false; - } - Networking::Packet Networking::popRequest() { std::lock_guard lock(mtx); if (!requestQueue.empty()) { @@ -107,8 +107,8 @@ namespace ttp2 { if (asn1_array2tree(packets_asn1_tab, &definitions, errorDescription) != ASN1_SUCCESS) { - std::wcout << "Error in sendMessage when loading asn1: " - << errorDescription << std::endl; + std::string errorDescriptionString = errorDescription; + logger->log(tablog::ERROR, "Error in sendMessage when loading asn1: " + errorDescriptionString); return -1; } @@ -119,15 +119,14 @@ namespace ttp2 { autoId++; } - std::string idString = std::to_string(id); - asn1_write_value(packet, "id", idString.c_str(), 0); + packet = Asn1Helpers::asn1EncodePayload(id, packet, "id"); if (std::holds_alternative(payload)) { // Write structure int status = asn1_write_value(packet, "payload", "standard", 0); if (status != ASN1_SUCCESS) { - std::wcout << "ASN1 set payload as standard failed!" << std::endl; + logger->log(tablog::ERROR, "ASN1 set payload as standard failed!"); } // Write contents @@ -139,7 +138,7 @@ namespace ttp2 { int status = asn1_write_value(packet, "payload", "file", 0); if (status != ASN1_SUCCESS) { - std::wcout << "ASN1 set payload as file failed!" << std::endl; + logger->log(tablog::ERROR, "ASN1 set payload as file failed!"); } // Write contents @@ -154,33 +153,65 @@ namespace ttp2 { std::shared_ptr table = std::get(payload).payload; std::shared_ptr buffer = tableToBuffer(table); - packet = Asn1Helpers::asn1EncodePayload(buffer->data(), buffer->size(), packet, "payload.file.payload"); + if (buffer->size() > 0) { + packet = Asn1Helpers::asn1EncodePayload(buffer->data(), buffer->size(), packet, "payload.file.payload"); + } + } else if (std::holds_alternative(payload)) { + // Write structure + int status = asn1_write_value(packet, "payload", "viewportRequest", 0); + + if (status != ASN1_SUCCESS) { + logger->log(tablog::ERROR, "ASN1 set payload as viewport request failed!"); + } + + // Write content + // X + int xStart = std::get(payload).xStart; + packet = Asn1Helpers::asn1EncodePayload(xStart, packet, "payload.viewportRequest.xStart"); + int xEnd = std::get(payload).xEnd; + packet = Asn1Helpers::asn1EncodePayload(xEnd, packet, "payload.viewportRequest.xEnd"); + + // Y + int yStart = std::get(payload).yStart; + packet = Asn1Helpers::asn1EncodePayload(yStart, packet, "payload.viewportRequest.yStart"); + int yEnd = std::get(payload).yEnd; + packet = Asn1Helpers::asn1EncodePayload(yEnd, packet, "payload.viewportRequest.yEnd"); } else if (std::holds_alternative(payload)) { // Write structure int status = asn1_write_value(packet, "payload", "viewport", 0); if (status != ASN1_SUCCESS) { - std::wcout << "ASN1 set payload as viewport failed!" << std::endl; + logger->log(tablog::ERROR, "ASN1 set payload as viewport failed!"); } // Write content // X int xStart = std::get(payload).xStart; packet = Asn1Helpers::asn1EncodePayload(xStart, packet, "payload.viewport.xStart"); - int xEnd = std::get(payload).xEnd; packet = Asn1Helpers::asn1EncodePayload(xEnd, packet, "payload.viewport.xEnd"); // Y int yStart = std::get(payload).yStart; packet = Asn1Helpers::asn1EncodePayload(yStart, packet, "payload.viewport.yStart"); - int yEnd = std::get(payload).yEnd; packet = Asn1Helpers::asn1EncodePayload(yEnd, packet, "payload.viewport.yEnd"); std::shared_ptr table = std::get(payload).payload; std::shared_ptr buffer = tableToBuffer(table); packet = Asn1Helpers::asn1EncodePayload(buffer->data(), buffer->size(), packet, "payload.viewport.payload"); + } else if (std::holds_alternative(payload)) { + // Write structure + int status = asn1_write_value(packet, "payload", "filter", 0); + if (status != ASN1_SUCCESS) { + logger->log(tablog::ERROR, "ASN1 set payload as filter failed!"); + } + + // Write content + std::string columnName = std::get(payload).columnName; + packet = Asn1Helpers::asn1EncodePayload(columnName, packet, "payload.filter.columnName"); + std::string regex = std::get(payload).regex; + packet = Asn1Helpers::asn1EncodePayload(regex, packet, "payload.filter.regex"); } int derLen = 0; @@ -188,8 +219,8 @@ namespace ttp2 { std::vector buffer(derLen); if (asn1_der_coding(packet, "", buffer.data(), &derLen, errorDescription) != ASN1_SUCCESS) { - std::wcout << "Error while encoding packet: " << errorDescription - << std::endl; + std::string errorDescriptionString = errorDescription; + logger->log(tablog::ERROR, "Error while encoding packet: " + errorDescriptionString); abort(); return -1; } @@ -217,10 +248,11 @@ namespace ttp2 { if (errno == EAGAIN || errno == EWOULDBLOCK) { break; } - // std::wcout << "Error while receiving!" << std::endl; + logger->log(tablog::ERROR, "Error while receiving bytes!"); return data; } else { - // std::wcout << "Socket closed!" << std::endl; + // logger->log(tablog::CRITICAL, "Socket closed!"); + disconnect(); return data; } } @@ -257,15 +289,7 @@ namespace ttp2 { if (asn1_der_decoding(&packet, derBuffer.data(), derLen, errorDescription) == ASN1_SUCCESS) { - unsigned char idBin[8]; - int idLen = sizeof(idBin); - if (asn1_read_value(packet, "id", idBin, &idLen) == ASN1_SUCCESS) { - long idValue = 0; - for (int i = 0; i < idLen; i++) { - idValue = (idValue << 8) | idBin[i]; - } - data.id = static_cast(idValue); - } + data.id = Asn1Helpers::asn1DecodePayloadInt(packet, "id"); char typeName[64]; int branchSize = sizeof(typeName); @@ -277,31 +301,46 @@ namespace ttp2 { data.payload = standard; } else if (typeNameString == "file") { - Networking::File file; - file.filePath = Asn1Helpers::asn1DecodePayloadString(packet, "payload.file.filePath"); - file.start = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.file.start"); - file.end = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.file.end"); + Networking::File file; + file.filePath = Asn1Helpers::asn1DecodePayloadString(packet, "payload.file.filePath"); + file.start = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.file.start"); + file.end = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.file.end"); + + std::vector buffer = Asn1Helpers::asn1DecodePayloadBuffer(packet, "payload.file.payload"); + // const uint8_t* bufferConst = buffer.data(); + file.payload = bufferToTable(buffer.data(), buffer.size()); - std::vector buffer = Asn1Helpers::asn1DecodePayloadBuffer(packet, "payload.file.payload"); - // const uint8_t* bufferConst = buffer.data(); - file.payload = bufferToTable(buffer.data(), buffer.size()); + data.payload = file; + } else if (typeNameString == "viewportRequest") { + Networking::ViewportRequest viewportRequest; - data.payload = file; + viewportRequest.xStart = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.viewportRequest.xStart"); + viewportRequest.xEnd = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.viewportRequest.xEnd"); + viewportRequest.yStart = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.viewportRequest.yStart"); + viewportRequest.yEnd = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.viewportRequest.yEnd"); + + data.payload = viewportRequest; } else if (typeNameString == "viewport") { - Networking::Viewport viewport; - viewport.xStart = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.viewport.xStart"); - viewport.xEnd = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.viewport.xEnd"); - viewport.yStart = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.viewport.yStart"); - viewport.yEnd = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.viewport.yEnd"); - std::vector buffer = Asn1Helpers::asn1DecodePayloadBuffer(packet, "payload.viewport.payload"); - viewport.payload = bufferToTable(buffer.data(), buffer.size()); - - data.payload = viewport; + Networking::Viewport viewport; + viewport.xStart = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.viewport.xStart"); + viewport.xEnd = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.viewport.xEnd"); + viewport.yStart = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.viewport.yStart"); + viewport.yEnd = Asn1Helpers::asn1DecodePayloadInt(packet, "payload.viewport.yEnd"); + std::vector buffer = Asn1Helpers::asn1DecodePayloadBuffer(packet, "payload.viewport.payload"); + viewport.payload = bufferToTable(buffer.data(), buffer.size()); + + data.payload = viewport; + } else if (typeNameString == "filter") { + Networking::Filter filter; + filter.columnName = Asn1Helpers::asn1DecodePayloadString(packet, "payload.filter.columnName"); + filter.regex = Asn1Helpers::asn1DecodePayloadString(packet, "payload.filter.regex"); + + data.payload = filter; } else { - std::wcout << "Error decoding payload: Unknown type!" << std::endl; + logger->log(tablog::ERROR, "Error decoding payload: Unknown type!"); } } else { - std::wcout << "Error decoding ASN1" << std::endl; + logger->log(tablog::ERROR, "Error decoding ASN1"); } asn1_delete_structure(&packet); @@ -355,6 +394,40 @@ namespace ttp2 { return result; } + Networking::PacketInfo Networking::peekResponse() { + return peekResponse(0); + } + + Networking::PacketInfo Networking::peekResponse(int index) { + Networking::PacketInfo packetInfo; + if (index > getResponseQueueSize() - 1 || index < 0) { + logger->log(tablog::ERROR, "Invalid peek request: " + std::to_string(index)); + return packetInfo; + } + + std::lock_guard lock(mtx); + packetInfo.id = responseQueue[index].id; + packetInfo.payloadType = responseQueue[index].payload; + return packetInfo; + } + + Networking::PacketInfo Networking::peekRequest() { + return peekRequest(0); + } + + Networking::PacketInfo Networking::peekRequest(int index) { + Networking::PacketInfo packetInfo; + if (index > getRequestQueueSize() - 1 || index < 0) { + logger->log(tablog::ERROR, "Invalid peek request: " + std::to_string(index)); + return packetInfo; + } + + std::lock_guard lock(mtx); + packetInfo.id = requestQueue[index].id; + packetInfo.payloadType = requestQueue[index].payload; + return packetInfo; + } + std::string Networking::getBroadcastIpAddress() { struct ifaddrs *ifaddr = nullptr; std::string broadcastIP; @@ -457,24 +530,68 @@ namespace ttp2 { arrow::Status status = streamWriter->WriteTable(*table); if (!status.ok()) { - std::wcout << "Something went wrong while writing the structure!" << std::endl; + logger->log(tablog::ERROR, "Something went wrong while writing the structure!"); + return nullptr; } streamWriter->Close(); arrow::Result> buffer = outputStream->Finish(); if (!buffer.ok()) { - std::wcout << "Something went wrong while converting table to buffer!" << std::endl; + logger->log(tablog::ERROR, "Something went wrong while converting table to buffer!"); + return nullptr; } - return *buffer; + return std::move(buffer).ValueUnsafe(); } std::shared_ptr Networking::bufferToTable(const uint8_t* rawData, int64_t dataSize) { - std::shared_ptr buffer = arrow::Buffer::Wrap(rawData, dataSize); + arrow::BufferBuilder bufferBuilder; + arrow::Status allocStatus = bufferBuilder.Resize(dataSize); + if (!allocStatus.ok()) { + logger->log(tablog::ERROR, "Buffer allocation failed in bufferToTable"); + return arrow::Table::Make(arrow::schema({}), std::vector>{}); + } + + // Make a physical copy so that the data isn't deleted. (That would lead to a shared_ptr with a table that points to no real data) + arrow::Status appendStatus = bufferBuilder.Append(reinterpret_cast(rawData), dataSize); + if (!appendStatus.ok()) { + logger->log(tablog::ERROR, "Failed to append raw data to buffer"); + return arrow::Table::Make(arrow::schema({}), std::vector>{}); + } + + std::shared_ptr buffer; + arrow::Status finishStatus = bufferBuilder.Finish(&buffer); + if (!finishStatus.ok()) { + logger->log(tablog::ERROR, "Failed to finish buffer building"); + return arrow::Table::Make(arrow::schema({}), std::vector>{}); + } + std::shared_ptr inputStream = std::make_shared(buffer); - std::shared_ptr stream_reader = *arrow::ipc::RecordBatchStreamReader::Open(inputStream); - std::shared_ptr table = *stream_reader->ToTable(); - return table; + + arrow::Result> streamReaderResult = arrow::ipc::RecordBatchStreamReader::Open(inputStream); + if (!streamReaderResult.ok()) { + logger->log(tablog::ERROR, "Open input stream failed in bufferToTable"); + return arrow::Table::Make(arrow::schema({}), std::vector>{}); + } + std::shared_ptr streamReader = std::move(streamReaderResult).ValueUnsafe(); + + arrow::Result> tableResult = streamReader->ToTable(); + if (!tableResult.ok()) { + logger->log(tablog::ERROR, "Create table failed in bufferToTable"); + return arrow::Table::Make(arrow::schema({}), std::vector>{}); + } + + return *tableResult; + } + + void Networking::disconnect() {} + + void Networking::configureLogger(std::string name) { + tablog::TablogRegistry* registry = &tablog::TablogRegistry::getInstance(); + std::shared_ptr logger = std::make_shared(); + logger->configure(name, true); + registry->registerLogger(name, logger); + this->logger = logger; } } diff --git a/lib/ttp2/src/packets.asn1 b/lib/ttp2/src/packets.asn1 index f4c228d..636e461 100644 --- a/lib/ttp2/src/packets.asn1 +++ b/lib/ttp2/src/packets.asn1 @@ -11,6 +11,13 @@ File ::= SEQUENCE { payload OCTET STRING } +ViewportRequest ::= SEQUENCE { + xStart INTEGER, + xEnd INTEGER, + yStart INTEGER, + yEnd INTEGER +} + Viewport ::= SEQUENCE { xStart INTEGER, xEnd INTEGER, @@ -19,12 +26,19 @@ Viewport ::= SEQUENCE { payload OCTET STRING } +Filter ::= SEQUENCE { + columnName OCTET STRING, + regex OCTET STRING +} + Packet ::= SEQUENCE { id INTEGER, payload CHOICE { standard [0] Standard, file [1] File, - viewport [2] Viewport + viewportRequest [2] ViewportRequest, + viewport [3] Viewport, + filter [4] Filter } } diff --git a/lib/ttp2/src/packets_asn1_tab.c b/lib/ttp2/src/packets_asn1_tab.c index 8e77560..a160d30 100644 --- a/lib/ttp2/src/packets_asn1_tab.c +++ b/lib/ttp2/src/packets_asn1_tab.c @@ -14,12 +14,20 @@ const asn1_static_node packets_asn1_tab[] = { { "start", 1073741827, NULL }, { "end", 1073741827, NULL }, { "payload", 7, NULL }, + { "ViewportRequest", 1610612741, NULL }, + { "xStart", 1073741827, NULL }, + { "xEnd", 1073741827, NULL }, + { "yStart", 1073741827, NULL }, + { "yEnd", 3, NULL }, { "Viewport", 1610612741, NULL }, { "xStart", 1073741827, NULL }, { "xEnd", 1073741827, NULL }, { "yStart", 1073741827, NULL }, { "yEnd", 1073741827, NULL }, { "payload", 7, NULL }, + { "Filter", 1610612741, NULL }, + { "columnName", 1073741831, NULL }, + { "regex", 7, NULL }, { "Packet", 536870917, NULL }, { "id", 1073741827, NULL }, { "payload", 536870930, NULL }, @@ -27,7 +35,11 @@ const asn1_static_node packets_asn1_tab[] = { { NULL, 2056, "0"}, { "file", 1610620930, "File"}, { NULL, 2056, "1"}, - { "viewport", 536879106, "Viewport"}, + { "viewportRequest", 1610620930, "ViewportRequest"}, { NULL, 2056, "2"}, + { "viewport", 1610620930, "Viewport"}, + { NULL, 2056, "3"}, + { "filter", 536879106, "Filter"}, + { NULL, 2056, "4"}, { NULL, 0, NULL } }; diff --git a/lib/ttp2/src/server_session_controller.cpp b/lib/ttp2/src/server_session_controller.cpp index 992f600..716ff1c 100644 --- a/lib/ttp2/src/server_session_controller.cpp +++ b/lib/ttp2/src/server_session_controller.cpp @@ -1,5 +1,7 @@ #include "../include/server_session_controller.h" +#include + #include #include #include @@ -10,9 +12,12 @@ #include namespace ttp2 { - ServerSessionController::ServerSessionController() {} + ServerSessionController::ServerSessionController() { + configureLogger("TTP2"); + } ServerSessionController::ServerSessionController(int serverSocket, int clientSocket) { + configureLogger("TTP2"); this->serverSocket = serverSocket; this->clientSocket = clientSocket; } @@ -20,13 +25,13 @@ namespace ttp2 { void ServerSessionController::networkingSession() { epollFd = epoll_create1(0); if (epollFd == -1) { - std::wcout << "Failed to create epoll!" << std::endl; + logger->log(tablog::ERROR, "Failed to create epoll!"); } clientEvent.events = EPOLLIN; clientEvent.data.fd = clientSocket; if (epoll_ctl(epollFd, EPOLL_CTL_ADD, clientSocket, &clientEvent) == -1) { - std::wcout << "Failed to set epoll_ctl for client!" << std::endl; + logger->log(tablog::ERROR, "Failed to set epoll_ctl for client!"); return; } @@ -61,10 +66,9 @@ namespace ttp2 { } void ServerSessionController::receiveRequestSession() { + const int MAX_EVENTS = 10; while (isConnected()) { - const int MAX_EVENTS = 10; struct epoll_event incomingEvents[MAX_EVENTS]; - int eventCount = epoll_wait(epollFd, incomingEvents, MAX_EVENTS, -1); for (int index = 0; index < eventCount; ++index) { @@ -72,18 +76,23 @@ namespace ttp2 { if (incomingEvents[index].events & (EPOLLHUP | EPOLLERR)) { sessionBuffers.erase(fd); close(fd); + disconnect(); continue; } if (incomingEvents[index].events & EPOLLIN) { - while (true) { - Packet packet = receiveMessage(fd); - if (packet.id == -1) { - break; - } - pushRequest(packet); + Packet packet = receiveMessage(fd); + if (packet.id == -1) { + continue; } + pushRequest(packet); } } } } + + void ServerSessionController::disconnect() { + std::lock_guard lock(mtx); + close(this->clientSocket); + connected = false; + } } diff --git a/test/client/CMakeLists.txt b/test/client/CMakeLists.txt index cd28d80..4be7657 100644 --- a/test/client/CMakeLists.txt +++ b/test/client/CMakeLists.txt @@ -11,5 +11,6 @@ add_executable(ttp2-client ${SOURCES}) target_link_libraries(ttp2-client PRIVATE ttp2) find_package(Arrow REQUIRED) target_link_libraries(ttp2-client PRIVATE Arrow::arrow_shared) +target_link_libraries(ttp2-client PRIVATE tablog) install(TARGETS ttp2-client DESTINATION bin) diff --git a/test/client/main.cpp b/test/client/main.cpp index 7566873..0a18d4d 100644 --- a/test/client/main.cpp +++ b/test/client/main.cpp @@ -1,7 +1,9 @@ #include "client_session_controller.h" #include +#include #include +#include #include #include #include @@ -42,6 +44,43 @@ int requestInt(const std::string& message) { } } +std::shared_ptr openCsvFile() { + std::string filePath { "" }; + do { + if (filePath.length() > 0 && !std::filesystem::exists(filePath)) { + std::wcout << "Incorrect filepath!" << std::endl; + } + filePath = requestString("(string) Filepath: "); + } while (!std::filesystem::exists(filePath)); + + arrow::io::IOContext ioContext = arrow::io::default_io_context(); + + arrow::Result> maybeFile = arrow::io::ReadableFile::Open(filePath); + std::shared_ptr fileInput = *maybeFile; + + arrow::csv::ReadOptions readOptions = arrow::csv::ReadOptions::Defaults(); + arrow::csv::ParseOptions parseOptions = arrow::csv::ParseOptions::Defaults(); + arrow::csv::ConvertOptions convertOptions = arrow::csv::ConvertOptions::Defaults(); + + arrow::Result> maybeReader = arrow::csv::TableReader::Make(ioContext, + fileInput, + readOptions, + parseOptions, + convertOptions); + if (!maybeReader.ok()) { + std::wcout << "Error while instantiating TableReader!" << std::endl; + } + std::shared_ptr reader = *maybeReader; + + arrow::Result> maybeTable = reader->Read(); + if (!maybeTable.ok()) { + std::wcout << "Error while read table from CSV file!" << std::endl; + } + std::shared_ptr table = *maybeTable; + return table; +} + + int main() { std::string ipAddress = requestString("Server ipv4 (string): "); int port = requestInt("Server port (int): "); @@ -66,13 +105,8 @@ int main() { clientSessionController->networkingSession(); }); - while (true) { - if (!clientSessionController->isConnected()) { - std::wcout << "Disconnect!" << std::endl; - break; - } - - int option = requestInt("Choose option\n(1) Send message\n(2) Read messages\n(3) Benchmark\n(4) Open file\n(5) Exit\nnumber: "); + while (clientSessionController->isConnected()) { + int option = requestInt("Choose option\n(1) Send message\n(2) Read messages\n(3) Benchmark\n(4) Open file\n(5) peek index\n(6) Viewport\n(7) Exit\nnumber: "); if (option == 1) { std::string payload = requestString("(string) Payload: "); @@ -101,7 +135,14 @@ int main() { std::wcout << "ID: " << packet.id << std::endl; std::wcout << file.payload->ToString().c_str() << std::endl; std::wcout << "---------------------" << std::endl; + } else if (std::holds_alternative(packet.payload)) { + Networking::Viewport viewport = std::get(packet.payload); + std::wcout << "------ Message Viewport------" << std::endl; + std::wcout << "ID: " << packet.id << std::endl; + std::wcout << viewport.payload->ToString().c_str() << std::endl; + std::wcout << "---------------------" << std::endl; } + } } else if (option == 3) { std::wcout << "~~~~~~ ~~~~~~ Benchmark ~~~~~~ ~~~~~~" << std::endl; @@ -151,41 +192,7 @@ int main() { std::wcout << "Invalid!" << std::endl; } } else if (option == 4) { - std::string filePath { "" }; - do { - if (filePath.length() > 0 && !std::filesystem::exists(filePath)) { - std::wcout << "Incorrect filepath!" << std::endl; - } - filePath = requestString("(string) Filepath: "); - } while (!std::filesystem::exists(filePath)); - - arrow::io::IOContext ioContext = arrow::io::default_io_context(); - - arrow::Result> maybeFile = arrow::io::ReadableFile::Open(filePath); - std::shared_ptr fileInput = *maybeFile; - - arrow::csv::ReadOptions readOptions = arrow::csv::ReadOptions::Defaults(); - arrow::csv::ParseOptions parseOptions = arrow::csv::ParseOptions::Defaults(); - arrow::csv::ConvertOptions convertOptions = arrow::csv::ConvertOptions::Defaults(); - - arrow::Result> maybeReader = arrow::csv::TableReader::Make(ioContext, - fileInput, - readOptions, - parseOptions, - convertOptions); - if (!maybeReader.ok()) { - std::wcout << "Error while instantiating TableReader!" << std::endl; - continue; - } - std::shared_ptr reader = *maybeReader; - - arrow::Result> maybeTable = reader->Read(); - if (!maybeTable.ok()) { - std::wcout << "Error while read table from CSV file!" << std::endl; - continue; - } - std::shared_ptr table = *maybeTable; - + std::shared_ptr table = openCsvFile(); ClientSessionController::Packet packet; ClientSessionController::File file; file.start = 0; @@ -196,12 +203,54 @@ int main() { std::wcout << "Done!" << std::endl; } else if (option == 5) { + int responseQueueSize = clientSessionController->getResponseQueueSize(); + int index = 0; + do { + std::wcout << "Peek index: 0 to " << responseQueueSize << " | -1 to exit" << std::endl; + index = requestInt("(int) index: "); + } while (index < -1 || index > responseQueueSize); + + if (index == -1) { + continue; + } + + ClientSessionController::PacketInfo packetInfo = clientSessionController->peekResponse(index); + std::wcout << "--- PacketInfo ---" << std::endl; + std::string payloadType = ""; + if (std::holds_alternative(packetInfo.payloadType)) + payloadType = "Standard"; + else if (std::holds_alternative(packetInfo.payloadType)) + payloadType = "File"; + else if (std::holds_alternative(packetInfo.payloadType)) + payloadType = "Viewport"; + else + payloadType = "Invalid"; + + std::wcout << "id: " << packetInfo.id << "\npacketType: " << payloadType.c_str() << std::endl; + std::wcout << "--- --- ---" << std::endl; + } else if (option == 6) { + std::shared_ptr table = openCsvFile(); + ClientSessionController::Packet packet; + ClientSessionController::Viewport viewport; + viewport.xStart = 0; + viewport.xEnd = table->num_rows(); + viewport.yStart = 0; + viewport.yEnd = table->num_columns(); + viewport.payload = table; + packet.payload = viewport; + clientSessionController->pushRequest(packet); + + std::wcout << "Done!" << std::endl; + } else if (option == 7) { clientSessionController->disconnect(); } else { std::wcout << "Invalid!" << std::endl; } } - networkThread.detach(); + + std::wcout << "Terminated!" << std::endl; + + networkThread.join(); return 0; } diff --git a/test/server/CMakeLists.txt b/test/server/CMakeLists.txt index 41bf1a7..637563a 100644 --- a/test/server/CMakeLists.txt +++ b/test/server/CMakeLists.txt @@ -11,5 +11,6 @@ add_executable(ttp2-server ${SOURCES}) target_link_libraries(ttp2-server PRIVATE ttp2) find_package(Arrow REQUIRED) target_link_libraries(ttp2-server PRIVATE Arrow::arrow_shared) +target_link_libraries(ttp2-server PRIVATE tablog) install(TARGETS ttp2-server DESTINATION bin)