Add local file persistence staging area for log events and player sco…

…res. Store data entries as newline-delimited JSON values.

src/game/Db.h

@@ -0,0 +1,141 @@
+#ifndef TRIPPIN_DB_H
+#define TRIPPIN_DB_H
+
+#include <sstream>
+#include <thread>
+#include "SDL.h"
+#include "nlohmann/json.hpp"
+#include "sprite/Files.h"
+#include "net/Channel.h"
+
+namespace trippin {
+    template<class T>
+    class Db {
+    public:
+        struct Event {
+            T val;
+            bool add;
+
+            bool operator==(const Event &e) const {
+                return e.val == val;
+            }
+        };
+
+        explicit Db(const char *name, int version, std::function<bool(const T&)> outFn) :
+                name(name),
+                outFn(outFn),
+                fileName(makeFileName(name, version)) {
+        }
+
+        void start() {
+            auto t1 = std::thread(&Db<T>::runFilePersistenceLoop, this);
+            t1.detach();
+            auto t2 = std::thread(&Db<T>::runDispatchLoop, this);
+            t2.detach();
+        }
+
+        void add(const T &val) {
+            inChannel.put({val, true});
+        }
+
+    private:
+        const char *name;
+        std::string fileName;
+        Channel<Event> inChannel;
+        Channel<T> outChannel;
+        std::function<bool(const T&)> outFn;
+
+        static std::string makeFileName(const char *name, int version) {
+            std::stringstream fileName;
+            fileName << name << "_journal_" << version << ".txt";
+            return fileName.str();
+        }
+
+        void load() {
+            auto contents = readPrefFile(fileName.c_str());
+            if (contents) {
+                SDL_Log("initializing from journal file, type=%s, filename=%s", name, fileName.c_str());
+                auto events = parseLines(*contents);
+                addInitialEvents(events);
+                SDL_Log("done initializing from journal file, type=%s, filename=%s", name, fileName.c_str());
+            } else {
+                SDL_Log("unable to initialize from journal file, type=%s, filename=%s", name, fileName.c_str());
+            }
+        }
+
+        static std::vector<Event> parseLines(const std::string &lines) {
+            std::istringstream iss(lines);
+            std::string line;
+            Event event;
+            std::vector<Event> events;
+            while (std::getline(iss, line)) {
+                try {
+                    auto j = nlohmann::json::parse(line);
+                    j.get_to(event);
+                    if (event.add) {
+                        events.push_back(event);
+                    } else {
+                        events.erase(std::remove(events.begin(), events.end(), event), events.end());
+                    }
+                } catch (nlohmann::json::exception &e) {
+                    SDL_Log("error occurred parsing json in journal, message=%s", e.what());
+                    break;
+                }
+            }
+            return events;
+        }
+
+        void addInitialEvents(const std::vector<Event> &events) {
+            std::stringstream ss;
+            for (auto &e: events) {
+                ss << nlohmann::json(e).dump() << "\n";
+            }
+            writePrefFile(fileName.c_str(), ss.str());
+            for (auto &e: events) {
+                outChannel.put(e.val);
+            }
+        }
+
+        void addNextEvent(Event &event) {
+            appendPrefFile(fileName.c_str(), nlohmann::json(event).dump() + "\n");
+            if (event.add) {
+                outChannel.put(event.val);
+            }
+        }
+
+        void runFilePersistenceLoop() {
+            SDL_Log("started file persistence thread, type=%s", name);
+            load();
+            unsigned long iter = 0;
+            while (true) {
+                auto event = inChannel.take();
+                if (!event) {
+                    break;
+                }
+                addNextEvent(*event);
+                iter++;
+                auto *op = event->add ? "add" : "ack";
+                SDL_Log("persisted event in journal, iter=%lu, type=%s, op=%s", iter, name, op);
+            }
+        }
+
+        void runDispatchLoop() {
+            SDL_Log("started dispatch thread, type=%s", name);
+            unsigned long iter = 0;
+            while (true) {
+                auto val = outChannel.take();
+                if (!val) {
+                    break;
+                }
+                while (!outFn(*val)) {
+                    std::this_thread::sleep_for(std::chrono::seconds(5));
+                }
+                inChannel.put({*val, false});
+                iter++;
+                SDL_Log("dispatched value, iter=%lu, type=%s", iter, name);
+            }
+        }
+    };
+}
+
+#endif

src/game/Game.cpp

@@ -5,7 +5,6 @@
 #include "SDL_syswm.h"
 #include "SDL_mixer.h"
 #include "Game.h"
-#include "net/DbSynchronizer.h"
 #include "UserInput.h"
 
 void trippin::Game::init() {
@@ -45,7 +44,7 @@ void trippin::Game::initAppId() {
 }
 
 void trippin::Game::initLogger() {
-    logger = std::make_unique<Logger>(*stagingArea, appId);
+    logger = std::make_unique<Logger>([this](const LogEvent& e){logDb->add(e);}, appId);
     SDL_DisplayMode displayMode;
     SDL_GetDisplayMode(0, 0, &displayMode);
     auto platform = SDL_GetPlatform();
@@ -65,18 +64,58 @@ void trippin::Game::initLogger() {
     logger->log(ss.str());
 }
 
+namespace trippin {
+    void to_json(nlohmann::json &j, const trippin::Db<Score>::Event &event) {
+        j["val"] = event.val;
+        j["add"] = event.add;
+    }
+
+    void from_json(const nlohmann::json &j, trippin::Db<Score>::Event &event) {
+        j.at("val").get_to(event.val);
+        j.at("add").get_to(event.add);
+    }
+
+    void to_json(nlohmann::json &j, const trippin::Db<LogEvent>::Event &event) {
+        j["val"] = event.val;
+        j["add"] = event.add;
+    }
+
+    void from_json(const nlohmann::json &j, trippin::Db<LogEvent>::Event &event) {
+        j.at("val").get_to(event.val);
+        j.at("add").get_to(event.add);
+    }
+};
+
 void trippin::Game::initDbSynchronizer() {
-    stagingArea = std::make_shared<StagingArea>();
-    Transport transport(
+    transport = std::make_unique<Transport>(
             configuration.db.host,
             configuration.db.port,
             configuration.version.major,
             configuration.highScores);
-    DbSynchronizer::startAddScoresThread(transport, stagingArea);
-    DbSynchronizer::startAddLogEventsThread(transport, stagingArea);
-    DbSynchronizer::startQueryScoresThread(std::move(transport), stagingArea);
+    stagingArea = std::make_unique<StagingArea>(*transport);
+    stagingArea->start();
     myScores = std::make_unique<MyScores>(configuration.version.major, 10);
     myScores->start();
+    scoreDb = std::make_unique<Db<Score>>("scores", configuration.version.major, [t = transport.get()](const Score &s) {
+        if (t->addScore(s)) {
+            SDL_Log("added score, id=%s, game=%d, name=%s, score=%d", s.id.c_str(), s.game, s.name.c_str(), s.score);
+            return true;
+        } else {
+            SDL_Log("failed to add score, id=%s, game=%d, name=%s, score=%d", s.id.c_str(), s.game, s.name.c_str(), s.score);
+            return false;
+        }
+    });
+    scoreDb->start();
+    logDb = std::make_unique<Db<LogEvent>>("logs", configuration.version.major, [t = transport.get()](const LogEvent &e) {
+        if (t->addLogEvent(e)) {
+            SDL_Log("sent log event, index=%d", e.index);
+            return true;
+        } else {
+            SDL_Log("failed to send log event, index=%d", e.index);
+            return false;
+        }
+    });
+    logDb->start();
 }
 
 void trippin::Game::initConfiguration() {
@@ -739,6 +778,7 @@ void trippin::Game::handle(UserInput::Event &event) {
         if (nameForm->nameEntered()) {
             Score sc{appId, gameId, score, nameForm->getName(), convertInputEvents()};
             stagingArea->addScore(sc);
+            scoreDb->add(sc);
             myScores->addScore(sc);
             state = State::START_MENU;
             titleMenu->reset();

src/game/Game.h

@@ -26,6 +26,7 @@
 #include "UserInput.h"
 #include "Random.h"
 #include "MyScores.h"
+#include "Db.h"
 
 namespace trippin {
     class Game {
@@ -90,8 +91,11 @@ namespace trippin {
         std::unique_ptr<SimpleOverlay> gameOverOverlay;
         std::unique_ptr<SimpleOverlay> levelsCompletedOverlay;
         std::unique_ptr<SimpleOverlay> trainingCompletedOverlay;
-        std::shared_ptr<StagingArea> stagingArea;
+        std::unique_ptr<StagingArea> stagingArea;
         std::unique_ptr<MyScores> myScores;
+        std::unique_ptr<Transport> transport;
+        std::unique_ptr<Db<Score>> scoreDb;
+        std::unique_ptr<Db<LogEvent>> logDb;
         Rect<int> scoreArea;
         Rect<int> flagsArea;
         RenderClock renderClock;

src/net/Channel.h

@@ -12,14 +12,15 @@ namespace trippin {
     public:
         std::optional<T> take() {
             std::unique_lock lock(mutex);
-            cv.wait(lock, [this]{ return closed || !queue.empty(); });
+            cv.wait(lock, [this] { return closed || !queue.empty(); });
             if (closed) {
                 return {};
             }
             auto e = queue.front();
             queue.pop();
             return e;
         }
+
         bool put(T elem) {
             {
                 std::lock_guard lock(mutex);
@@ -31,13 +32,15 @@ namespace trippin {
             cv.notify_one();
             return true;
         }
+
         void close() {
             {
                 std::unique_lock lk(mutex);
                 closed = true;
             }
             cv.notify_all();
         }
+
     private:
         std::queue<T> queue;
         std::mutex mutex;