renamed to pierog, extracted socket and file writers

03_Perun_Lua_Wrapper/CMakeLists.txt

@@ -14,5 +14,5 @@ if (CMAKE_CXX_COMPILER_VERSION VERSION_GREATER_EQUAL 19.12.25835)
 endif()
 
 add_subdirectory(lua-5.1.5)
-add_subdirectory(perun)
+add_subdirectory(pierog)
 add_subdirectory(experimental)

03_Perun_Lua_Wrapper/experimental/CMakeLists.txt

@@ -9,10 +9,13 @@ set (EXPERIMENTAL_SOURCES
 
 add_executable(experimental ${EXPERIMENTAL_SOURCES})
 
+include_directories(${pierog_SOURCE_DIR})
+
 target_link_libraries(
-        lua-5.1.5
+        experimental
         ws2_32
+        pierog
 )
 
-target_include_directories (pierog PUBLIC)
+target_include_directories (pierog PUBLIC pierog)
-#set_target_properties(main PROPERTIES OUTPUT_NAME "perun")
+#set_target_properties(main PROPERTIES OUTPUT_NAME "pierog")

03_Perun_Lua_Wrapper/lua-5.1.5/src/lcode.c

@@ -44,7 +44,7 @@ void luaK_nil (FuncState *fs, int from, int n) {
       if (GET_OPCODE(*previous) == OP_LOADNIL) {
         int pfrom = GETARG_A(*previous);
         int pto = GETARG_B(*previous);
-        if (pfrom <= from && from <= pto+1) {  /* can connect both? */
+        if (pfrom <= from && from <= pto+1) {  /* can _connect both? */
           if (from+n-1 > pto)
             SETARG_B(*previous, from+n-1);
           return;

03_Perun_Lua_Wrapper/perun/CmakeLists.txt

@@ -1,22 +0,0 @@
-cmake_minimum_required(VERSION 3.17)
-project(pierog)
-
-set(CMAKE_CXX_STANDARD 20)
-
-set (PIEROG_DLL_SOURCES
-    "src/library.h" 
-    "src/library.cpp" 
-    "src/Connection.h" 
-    "src/Connection.cpp"
-    )
-
-include (GenerateExportHeader)
-add_library(pierog SHARED ${PIEROG_DLL_SOURCES})
-
-target_link_libraries(
-        pierog
-        lua-5.1.5
-        ws2_32
-)
-
-target_include_directories (pierog PUBLIC)

03_Perun_Lua_Wrapper/perun/src/Connection.cpp

@@ -1,164 +0,0 @@
-#include "Connection.h"
-
-#include <utility>
-#include <filesystem>
-
-SocketWrapper::SocketWrapper(std::string logPath,
-                             std::string host,
-                             const int port): path(std::move(logPath)), tcpHost(std::move(host)), tcpPort(port){
-	this->tcpSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
-
-    startNewRecording();
-}
-
-SocketWrapper::~SocketWrapper() {
-}
-
-void SocketWrapper::tcpConnect() {
-    // Create socket address object from TCP port and host
-    SOCKADDR_IN socketAddress;
-    socketAddress.sin_family = AF_INET;
-    socketAddress.sin_port = htons(u_short(this->tcpPort));
-    socketAddress.sin_addr.s_addr = *((unsigned long*)gethostbyname(this->tcpHost.c_str())->h_addr);
-
-    if (connect(tcpSocket, (sockaddr*)&socketAddress, sizeof(SOCKADDR_IN)) == 0) {
-        this->connectionState = CONNECTED;
-        this->flagReconnected = 1;
-    }
-}
-
-void SocketWrapper::startNewRecording() {
-    if(outputFile != nullptr && outputFile->is_open() && outputFile->good()) {
-        outputFile->flush();
-        outputFile->close();
-        delete outputFile;
-    }
-
-    auto const now = std::chrono::system_clock::now();
-    auto const gmt = std::chrono::locate_zone("Etc/GMT");
-    auto const filename = std::format("pierog.{:%FT_%H%M%S}.log", std::chrono::zoned_time{gmt, floor<std::chrono::milliseconds>(now)});
-
-    std::filesystem::path dir(this->path);
-    std::filesystem::path file(filename);
-    auto fullPath = (dir / file).string();
-
-    outputFile = new std::ofstream(fullPath, std::ofstream::app | std::ios::out);
-
-    // clean the network buffer if nothing was ever sent
-    if(sentCounter > 0) {
-        mutexLock.lock();
-        dataBuffer.clear();
-        sendQueue.clear();
-        mutexLock.unlock();
-    }
-}
-
-void SocketWrapper::createConnection() {
-    // TCP connection - ConnectTo
-
-    tcpConnect();
-
-    std::thread thread_object([this]() {
-
-        bool nothingToSend = false;
-        while (shouldRun) {
-            if (connectionState == CONNECTED && mutexLock.try_lock()) {
-                if (sendQueue.empty()) {
-                    nothingToSend = true;
-                } else {
-                    // Payload in queue
-                    auto payload = sendQueue.front();
-                    int bytesSent = send(tcpSocket, payload->c_str(), payload->length(), 0);
-
-                    if (bytesSent == payload->length()) {
-                        // All payload was sent
-                        sendQueue.pop_front();
-                        delete payload;
-                        sentCounter += bytesSent;
-                    } else {
-                        // Remaining paylad
-                        if (bytesSent > 0) {
-                            // Send remaining bytes
-                            auto shortened = payload->substr(bytesSent, payload->length() - bytesSent);
-                            sendQueue.pop_front();
-                            sendQueue.push_front(&shortened);
-                            delete payload;
-                            sentCounter += bytesSent;
-                        } else {
-                            // Payload was not sent - handle error
-                            switch (WSAGetLastError()) {
-                                // Connection was reset
-                                case WSAECONNRESET:
-                                // Connection aborted
-                                case WSAECONNABORTED:
-                                // Connection was closed
-                                case WSAESHUTDOWN:
-                                    connectionState = DISCONNECTED;
-                                    reconnect();
-                            }
-                        }
-                    }
-                }
-                mutexLock.unlock();
-            } else {
-                // Not connected
-                reconnect();
-            }
-            // sleep longer if nothing to send
-            if (nothingToSend) { Sleep(100); } else { Sleep(10); }
-        }
-        });
-    thread_object.detach(); // Detach TCP thread from main thread
-}
-
-void SocketWrapper::reconnect() {
-    if (connectionState == DISCONNECTED) {
-        disconnect();
-        tcpSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); // Reset socket
-    }
-
-    tcpConnect();
-}
-
-void SocketWrapper::disconnect() {
-    // TCP connection - Disconnect
-	closesocket(tcpSocket);
-    connectionState = DISCONNECTED;
-}
-
-void SocketWrapper::enqueueForSending(std::string* payload) {
-    if(outputFile == nullptr) {
-        startNewRecording();
-    }
-
-    if(outputFile != nullptr) {
-        outputFile->write(payload->c_str(), payload->length());
-        if(payload->length() > 50) {
-            outputFile->flush();
-        }
-    }
-    if (mutexLock.try_lock()) {
-        while (!dataBuffer.empty()) {
-            // Shift buffer to queue
-            sendQueue.push_back(dataBuffer.front());
-            dataBuffer.pop_front();
-        }
-        sendQueue.push_back(payload);  
-        mutexLock.unlock(); 
-    }
-    else {
-        dataBuffer.push_back(payload);
-    }
-}
-
-int SocketWrapper::getAndResetReconnected() {
-    int result = this->flagReconnected;
-
-    this->flagReconnected = 0;
-
-    return result;
-}
-
-int SocketWrapper::getFlagConnected() {
-    return this->connectionState;
-}

03_Perun_Lua_Wrapper/perun/src/Connection.h

@@ -1,56 +0,0 @@
-#ifndef PERUN_CONNECTION_H
-#define PERUN_CONNECTION_H
-
-#include "winsock.h"
-
-#include <string>
-#include <queue>
-#include <mutex>
-#include <fstream>
-
-#include "library.h"
-
-enum enumConnectionState {
-	DISCONNECTED,
-	CONNECTED,
-};
-
-class SocketWrapper {
-public:
-    SocketWrapper(std::string logPath,
-                  std::string host,
-                  const int port);
-
-    ~SocketWrapper();
-
-	void disconnect();
-	void createConnection();
-    void startNewRecording();
-	void enqueueForSending(std::string* payload);
-	
-	int getAndResetReconnected();
-	int getFlagConnected();
-
-private:
-	SOCKET tcpSocket;
-    const std::string path;
-	const std::string tcpHost = "localhost";
-	const int tcpPort = 0;
-
-	int flagReconnected = 0;
-	volatile enumConnectionState connectionState = DISCONNECTED;
-    std::atomic<long long> sentCounter;
-    std::atomic<boolean> shouldRun = true;
-
-	std::deque<std::string*> dataBuffer;
-	std::deque<std::string*> sendQueue;
-
-	std::mutex mutexLock;
-    std::ofstream * outputFile = nullptr;
-
-	void reconnect();
-	void tcpConnect();
-};
-
-
-#endif

03_Perun_Lua_Wrapper/pierog/CmakeLists.txt

@@ -0,0 +1,25 @@
+cmake_minimum_required(VERSION 3.17)
+project(pierog)
+
+set(CMAKE_CXX_STANDARD 20)
+
+set(PIEROG_DLL_SOURCES
+        "src/library.h"
+        "src/library.cpp"
+        src/RotatingFileOutput.cpp
+        src/SocketOutput.cpp
+        src/DataDistributor.cpp
+        src/RotatingFileOutput.h
+        src/SocketOutput.h
+        src/DataDistributor.h)
+
+include(GenerateExportHeader)
+add_library(pierog SHARED ${PIEROG_DLL_SOURCES})
+
+target_link_libraries(
+        pierog
+        lua-5.1.5
+        ws2_32
+)
+
+target_include_directories(pierog PUBLIC ${PIEROG_DLL_SOURCES})

03_Perun_Lua_Wrapper/pierog/src/DataDistributor.cpp

@@ -0,0 +1,102 @@
+#include "DataDistributor.h"
+
+#include <utility>
+
+DataDistributor::DataDistributor(std::string logPath,
+                                 std::string host,
+                                 int port) {
+
+    fileOutput = new RotatingFileOutput(std::move(logPath));
+    socketOutput = new SocketOutput(std::move(host), port);
+}
+
+DataDistributor::~DataDistributor() = default;
+
+void DataDistributor::start() {
+    if(running) {
+        return;
+    }
+
+    std::thread thread_object([this]() {
+        bool anythingToSend = true;
+        long KEEP_ALIVE = 1000;
+        auto now = std::chrono::system_clock::now();
+
+        while(shouldRun) {
+            if(lock.try_lock()) {
+                if(sendQueue.empty()) {
+                    auto delay = std::chrono::duration_cast<std::chrono::milliseconds>(now - lastSent);
+                    if(delay.count() > KEEP_ALIVE) {
+                        sendQueue.push_front(new std::string(" "));
+                    } else {
+                        anythingToSend = false;
+                    }
+                } else {
+                    auto payload = sendQueue.front();
+                    int sent = socketOutput->write(payload);
+
+                    if(sent > 0) {
+                        everSentViaSocket = true;
+                        sendQueue.pop_front();
+                        if(sent == payload->length()) {
+                            delete payload;
+                        } else {
+                            auto shortened = payload->substr(sent, payload->length() - sent);
+                            sendQueue.push_front(&shortened);
+                            delete payload;
+                        }
+                        lastSent = std::chrono::system_clock::now();
+                    } else {
+                        // failed to send
+                    }
+                }
+
+                lock.unlock();
+            }
+
+            if(anythingToSend) {
+                Sleep(10);
+            } else {
+                Sleep(100);
+            }
+        }
+    });
+    thread_object.detach();
+
+    shouldRun = true;
+    running = true;
+}
+
+void DataDistributor::stop() {
+    shouldRun = false;
+}
+
+void DataDistributor::enqueueForSending(std::string *payload) {
+    fileOutput->write(payload);
+    if (lock.try_lock()) {
+        while (!dataBuffer.empty()) {
+            // Shift buffer to queue
+            sendQueue.push_back(dataBuffer.front());
+            dataBuffer.pop_front();
+        }
+        sendQueue.push_back(payload);
+        lock.unlock();
+    }
+    else {
+        dataBuffer.push_back(payload);
+    }
+}
+
+void DataDistributor::markNewRecording() {
+    fileOutput->markNewRecording();
+    if(!everSentViaSocket) {
+        lock.lock();
+        sendQueue.clear();
+        dataBuffer.clear();
+        lock.unlock();
+    }
+}
+
+int DataDistributor::isConnected() {
+    return socketOutput->isConnected();
+}

03_Perun_Lua_Wrapper/pierog/src/DataDistributor.h

@@ -0,0 +1,40 @@
+#ifndef PARENT_DATADISTRIBUTOR_H
+#define PARENT_DATADISTRIBUTOR_H
+
+#include <string>
+#include <queue>
+#include "SocketOutput.h"
+#include "RotatingFileOutput.h"
+
+class DataDistributor {
+public:
+    DataDistributor(std::string logPath,
+                    std::string host,
+                    int port);
+
+    virtual ~DataDistributor();
+
+    void enqueueForSending(std::string* payload);
+    void markNewRecording();
+
+    void start();
+    void stop();
+
+    int isConnected();
+
+private:
+    std::atomic<boolean> shouldRun = true;
+    std::atomic<boolean> running = false;
+    std::atomic<boolean> everSentViaSocket = false;
+    std::deque<std::string*> dataBuffer;
+    std::deque<std::string*> sendQueue;
+    std::mutex lock;
+
+    std::chrono::time_point<std::chrono::system_clock> lastSent = std::chrono::system_clock::now();
+
+    SocketOutput* socketOutput;
+    RotatingFileOutput* fileOutput;
+};
+
+
+#endif //PARENT_DATADISTRIBUTOR_H

03_Perun_Lua_Wrapper/pierog/src/RotatingFileOutput.cpp

@@ -0,0 +1,41 @@
+#include "RotatingFileOutput.h"
+
+RotatingFileOutput::~RotatingFileOutput() = default;
+
+RotatingFileOutput::RotatingFileOutput(std::string outputPath): path(std::move(outputPath)) {
+
+}
+
+void RotatingFileOutput::markNewRecording() {
+    if(outputFile != nullptr && outputFile->good()) {
+        outputFile->flush();
+        outputFile->close();
+        delete outputFile;
+        outputFile = nullptr;
+    }
+}
+
+void RotatingFileOutput::write(std::string *payload) {
+    if (outputFile == nullptr) {
+        std::string fileName = generateFileName();
+        outputFile = new std::ofstream(fileName, std::ofstream::app | std::ios::out);
+    }
+
+    outputFile->write(payload->c_str(), payload->length());
+    bytesWritten += (long) payload->length();
+    if (payload->length() > 50) {
+        outputFile->flush();
+    }
+}
+
+std::string RotatingFileOutput::generateFileName() {
+    auto const now = std::chrono::system_clock::now();
+    auto const gmt = std::chrono::locate_zone("Etc/GMT");
+    auto const filename = std::format("pierog.{:%FT_%H%M%S}.log",
+                                      std::chrono::zoned_time{gmt, floor<std::chrono::milliseconds>(now)});
+
+    std::filesystem::path dir(this->path);
+    std::filesystem::path file(filename);
+    return (dir / file).string();
+}
+

03_Perun_Lua_Wrapper/pierog/src/RotatingFileOutput.h

@@ -0,0 +1,27 @@
+#ifndef PARENT_ROTATINGFILEOUTPUT_H
+#define PARENT_ROTATINGFILEOUTPUT_H
+
+#include <mutex>
+#include <fstream>
+#include <iostream>
+#include <filesystem>
+
+class RotatingFileOutput {
+public:
+    RotatingFileOutput(std::string outputPath);
+    virtual ~RotatingFileOutput();
+
+    void markNewRecording();
+    void write(std::string* payload);
+
+private:
+    const std::string path;
+
+    std::atomic<long long> bytesWritten;
+    std::ofstream * outputFile = nullptr;
+
+    std::string generateFileName();
+};
+
+
+#endif //PARENT_ROTATINGFILEOUTPUT_H

03_Perun_Lua_Wrapper/pierog/src/SocketOutput.cpp

@@ -0,0 +1,60 @@
+#include "SocketOutput.h"
+
+SocketOutput::SocketOutput(std::string host,
+                           const int port): tcpHost(std::move(host)), tcpPort(port) {
+
+    address = new SOCKADDR_IN;
+    address->sin_family = AF_INET;
+    address->sin_port = htons(u_short(this->tcpPort));
+    address->sin_addr.s_addr = *((unsigned long*)gethostbyname(this->tcpHost.c_str())->h_addr);
+}
+
+SocketOutput::~SocketOutput() = default;
+
+int SocketOutput::write(std::string *payload) {
+
+    if(!(isConnected() || _connect())) {
+        return 0;
+    }
+
+    int bytesSent = send(tcpSocket, payload->c_str(), payload->length(), 0);
+    if(bytesSent == payload->length()) {
+        return 0;
+    }
+
+    if(bytesSent > 0) {
+        return bytesSent;
+    } else {
+        switch (WSAGetLastError()) {
+            case WSAECONNRESET:     // Connection reset
+            case WSAECONNABORTED:   // Connection aborted
+            case WSAESHUTDOWN:      // Connection closed
+                disconnect();
+        }
+        return 0;
+    }
+}
+
+bool SocketOutput::isConnected() {
+    return connectionState == CONNECTED;
+}
+
+bool SocketOutput::_connect() {
+    if(tcpSocket == INVALID_SOCKET) {
+        tcpSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+    }
+    if(connectionState != CONNECTED) {
+        if (connect(tcpSocket, (sockaddr *) address, sizeof(SOCKADDR_IN)) == 0) {
+            connectionState = CONNECTED;
+            return true;
+        }
+    }
+
+    return false;
+}
+
+void SocketOutput::disconnect() {
+    closesocket(tcpSocket);
+    tcpSocket = INVALID_SOCKET;
+    connectionState = DISCONNECTED;
+}

03_Perun_Lua_Wrapper/pierog/src/library.cpp

@@ -1,8 +1,9 @@
 #include "library.h"
+#include "DataDistributor.h"
 #include <format>
 #include <filesystem>
 
-static SocketWrapper * tcpConnection = nullptr;
+static DataDistributor * dataDistributor = nullptr;
 
 static std::string *startingDelimiter = nullptr;
 static std::string *endingDelimiter = nullptr;
@@ -15,15 +16,15 @@ static int valuesToReturn(int input) {
 
 static int appStartHook(lua_State* luaState) {
     // Starting the app - prepare
-    if(tcpConnection == nullptr) {
+    if(dataDistributor == nullptr) {
 
         int i = 1;
         const std::string path = std::string(lua_tolstring(luaState, i++, 0));
         const std::string host = std::string(lua_tolstring(luaState, i++, 0));
         const int port = (int) lua_tointeger(luaState, i++);
 
-        tcpConnection = new SocketWrapper(path, host, port);
+        dataDistributor = new DataDistributor(path, host, port);
-        tcpConnection->createConnection();
+        dataDistributor->start();
     }
 
     lua_pushinteger(luaState, 1);       // First return value: confirmation that app was started
@@ -34,8 +35,8 @@ static int appStartHook(lua_State* luaState) {
 static int appEndHook(lua_State* luaState) {
     // Closing the app - clean up
 
-    if(tcpConnection != nullptr) {
+    if(dataDistributor != nullptr) {
-        tcpConnection->disconnect();
+        dataDistributor->stop();
     }
 
     return valuesToReturn(0);
@@ -52,8 +53,8 @@ static int markMissionStart(lua_State* luaState) {
 
     std::string missionHash = std::string(lua_tolstring(luaState, 1, 0));
 
-    if(tcpConnection != nullptr && missionHash ==lastObservedHash) {
+    if(dataDistributor != nullptr && missionHash != lastObservedHash) {
-        tcpConnection->startNewRecording();
+        dataDistributor->markNewRecording();
         lastObservedHash = missionHash;
     }
 
@@ -63,25 +64,22 @@ static int markMissionStart(lua_State* luaState) {
 static int tcpSend(lua_State* luaState) {
     // Send frame over TCP socket 
 
-    if(tcpConnection != nullptr) {
+    if(dataDistributor != nullptr) {
         if(startingDelimiter != nullptr) {
-            tcpConnection->enqueueForSending(new std::string(*startingDelimiter));
+            dataDistributor->enqueueForSending(new std::string(*startingDelimiter));
         }
-        tcpConnection->enqueueForSending(new std::string(lua_tolstring(luaState, 1, 0)));
+        dataDistributor->enqueueForSending(new std::string(lua_tolstring(luaState, 1, 0)));
         if(endingDelimiter != nullptr) {
-            tcpConnection->enqueueForSending(new std::string(*endingDelimiter));
+            dataDistributor->enqueueForSending(new std::string(*endingDelimiter));
         }
 
         lua_pushinteger(luaState,
-                        tcpConnection->getFlagConnected());     // First return value: information if there is TCP connection
+                        dataDistributor->isConnected());     // First return value: information if there is TCP connection
-        lua_pushinteger(luaState,
-                        tcpConnection->getAndResetReconnected());   // Second return value: information if there was recent reconnection to TCP server
     } else {
         lua_pushinteger(luaState, 0);
-        lua_pushinteger(luaState, 0);
     }
 
-    return valuesToReturn(2);
+    return valuesToReturn(1);
 }
 
 extern "C" int __declspec(dllexport) luaopen_pierog(lua_State * L) {

03_Perun_Lua_Wrapper/pierog/src/library.h

@@ -12,6 +12,5 @@ extern "C" {
 #include <fstream>
 #include <chrono>
 #include <queue>
-#include "Connection.h"
 
 #endif //PERUN_LIBRARY_H