first commit

src/Services/AudioStreamService.h

@@ -0,0 +1,102 @@
+#pragma once
+
+#include <sio_client.h>
+#include <atomic>
+#include <spdlog/spdlog.h>
+#include <memory>
+#include <utility>
+
+namespace snoop {
+
+class AudioStreamService {
+    std::shared_ptr<sio::client> m_client;
+    std::string m_guid;
+    std::atomic<bool> m_isConnected = false;
+    std::atomic<bool> m_isInStreaming = false;
+    std::vector<char> m_audioBuffer;
+    std::mutex m_bufferMutex;
+    const unsigned long long int FLUSH_PERIOD = 5000;
+    const std::vector<char> PACKET_DELIMITER = {
+        static_cast<char>(0xFF),
+        static_cast<char>(0xFE),
+        static_cast<char>(0xFD),
+        static_cast<char>(0xFC)
+    };
+    unsigned long long int m_flushedAt = 0;
+
+public:
+    explicit AudioStreamService( std::shared_ptr<sio::client> client, std::string guid ) :
+        m_client( std::move( client ) ),
+        m_guid( std::move( guid ) ) {
+        SetupEventListeners();
+    }
+
+    ~AudioStreamService() {
+        this->m_isConnected = false;
+        this->m_isInStreaming = false;
+    }
+
+    void SendAudioData( const char* input, size_t size ) {
+        if( !this->m_isConnected || !this->m_isInStreaming ) {
+            return;
+        }
+
+        std::lock_guard lock( m_bufferMutex );
+        this->m_audioBuffer.insert(m_audioBuffer.end(), PACKET_DELIMITER.begin(), PACKET_DELIMITER.end());
+        this->m_audioBuffer.insert( m_audioBuffer.end(), input, input + size );
+
+        auto now = std::chrono::system_clock::now();
+        auto currentTime = std::chrono::duration_cast<std::chrono::milliseconds>( now.time_since_epoch() ).count();
+        if( currentTime >= this->m_flushedAt + FLUSH_PERIOD ) {
+            FlushBuffer();
+        }
+    }
+
+private:
+    void SetupEventListeners() {
+        this->m_client->set_open_listener( [this]() {
+            spdlog::info( "Connected to server" );
+            this->m_client->socket( "/livestream" )->emit( "register_device", m_guid );
+            this->m_isConnected = true;
+        } );
+
+        this->m_client->set_close_listener( [this]( sio::client::close_reason const& reason ) {
+            this->m_isConnected = false;
+            this->m_isInStreaming = false;
+            spdlog::info( "Disconnected from server" );
+        } );
+
+        this->m_client->set_fail_listener( []() {
+            spdlog::info( "Failed to connect to server" );
+        } );
+
+        this->m_client->socket( "/livestream" )->on( "start_streaming", [this]( sio::event& ev ) {
+            spdlog::info( "Start streaming command received" );
+            this->m_isInStreaming = true;
+            auto now = std::chrono::system_clock::now();
+            this->m_flushedAt = std::chrono::duration_cast<std::chrono::milliseconds>( now.time_since_epoch() ).count();
+        } );
+
+        this->m_client->socket( "/livestream" )->on( "stop_streaming", [this]( sio::event& ev ) {
+            spdlog::info( "Stop streaming command received" );
+            this->m_isInStreaming = false;
+            std::lock_guard lock( this->m_bufferMutex );
+            this->m_audioBuffer.clear();
+        } );
+    }
+
+    void FlushBuffer() {
+        if( this->m_audioBuffer.empty() ) {
+            return;
+        }
+
+        this->m_client->socket( "/livestream" )->emit( "audio_data",
+                                                       std::make_shared<std::string>( this->m_audioBuffer.data(), this->m_audioBuffer.size() )
+            );
+        this->m_audioBuffer.clear();
+        auto now = std::chrono::system_clock::now();
+        this->m_flushedAt = std::chrono::duration_cast<std::chrono::milliseconds>( now.time_since_epoch() ).count();
+    }
+};
+
+}

src/Services/AudioWriterService.h

@@ -0,0 +1,181 @@
+#pragma once
+
+#include <memory>
+#include <utility>
+#include <thread>
+#include <atomic>
+#include <chrono>
+#include <httplib.h>
+#include <mutex>
+
+#include "AudioWriters/OggAudioWriter.h"
+#include "ConfigService.h"
+
+
+namespace snoop {
+
+class AudioWriterService {
+    std::shared_ptr<ConfigService> m_configService;
+    std::shared_ptr<OggAudioWriter> m_oggWriter;
+    std::string m_destinationDirectoryPath;
+    std::string m_queueDirectoryPath;
+    std::thread m_writingThread;
+    std::thread m_uploadThread;
+    std::mutex m_fetchFilePathsMutex;
+    unsigned long long int m_currentRecordStartedAt = 0;
+    std::string m_currentRecordFilePath;
+    std::atomic<bool> m_isIntermission = false;
+
+public:
+    explicit AudioWriterService( std::shared_ptr<ConfigService> configService, std::string destinationDirectoryPath ) :
+        m_configService( std::move( configService ) ),
+        m_destinationDirectoryPath( std::move( destinationDirectoryPath ) ),
+        m_oggWriter( std::make_shared<OggAudioWriter>( 48000, 1 ) ) {
+        if( !this->m_destinationDirectoryPath.empty() && !this->m_destinationDirectoryPath.ends_with( "/" ) ) {
+            this->m_destinationDirectoryPath.append( "/" );
+        }
+        this->m_queueDirectoryPath = this->m_destinationDirectoryPath + "queue/";
+        std::filesystem::create_directories( this->m_queueDirectoryPath );
+        this->MoveToQueueUncompletedRecords();
+        this->m_writingThread = std::thread( [this]() {
+            this->WritingThread();
+        } );
+        this->m_uploadThread = std::thread( [this]() {
+            this->UploadThread();
+        } );
+        spdlog::info( "AudioWriterService::AudioWriterService()" );
+    }
+
+    void WriteAudioData( const char* data, size_t size, size_t frames ) {
+        this->m_oggWriter->Write( data, size, frames );
+    }
+
+    ~AudioWriterService() {
+        this->m_isIntermission = true;
+        this->m_writingThread.join();
+        this->m_uploadThread.join();
+    }
+
+private:
+    void MoveToQueueUncompletedRecords() {
+        std::vector<std::filesystem::path> files;
+        for( const auto& entry: std::filesystem::directory_iterator( this->m_destinationDirectoryPath ) ) {
+            files.push_back( entry.path() );
+        }
+        for( const auto& file: files ) {
+            if( file.filename().string() != "queue" ) {
+                spdlog::info( "Move uncompleted record {} to queue", file.filename().string() );
+                this->MoveToUploadQueue( file.string() );
+            }
+        }
+    }
+
+    void WritingThread() {
+        auto now = std::chrono::system_clock::now();
+        this->m_currentRecordStartedAt = std::chrono::duration_cast<std::chrono::milliseconds>( now.time_since_epoch() ).count();
+        this->m_currentRecordFilePath = this->m_destinationDirectoryPath + std::to_string( this->m_currentRecordStartedAt );
+        this->m_oggWriter->StartWriting( this->m_currentRecordFilePath );
+
+
+        while( !m_isIntermission ) {
+            now = std::chrono::system_clock::now();
+            auto currentRecordDuration = std::chrono::duration_cast<std::chrono::milliseconds>( now.time_since_epoch() ).count() -
+                this->m_currentRecordStartedAt;
+            if( currentRecordDuration >= this->m_configService->GetRecordingDuration() ) {
+                this->m_oggWriter->StopWriting();
+                this->MoveToUploadQueue( this->m_currentRecordFilePath );
+                this->m_currentRecordStartedAt = std::chrono::duration_cast<std::chrono::milliseconds>( now.time_since_epoch() ).count();
+                this->m_currentRecordFilePath = this->m_destinationDirectoryPath + std::to_string( this->m_currentRecordStartedAt );
+                this->m_oggWriter->StartWriting( this->m_currentRecordFilePath );
+            }
+            std::this_thread::sleep_for( std::chrono::milliseconds( 1000 ) );
+        }
+
+        this->m_oggWriter->StopWriting();
+        this->MoveToUploadQueue( this->m_currentRecordFilePath );
+        // TODO: Move to upload queue
+    }
+
+    void MoveToUploadQueue( const std::string& filePath ) {
+        spdlog::info( "AudioWriterService::MoveToUploadQueue( {} )", filePath );
+        std::lock_guard lock( this->m_fetchFilePathsMutex );
+        auto now = std::chrono::system_clock::now();
+        auto recordStoppedAt = std::chrono::duration_cast<std::chrono::milliseconds>( now.time_since_epoch() ).count();
+        if( std::filesystem::exists( filePath ) ) {
+            auto fileName = std::filesystem::path( filePath ).filename().string() + "-" + std::to_string( recordStoppedAt );
+            std::filesystem::rename( filePath, m_queueDirectoryPath + fileName );
+        }
+    }
+
+    void UploadThread() {
+        while (!m_isIntermission) {
+            std::vector<std::filesystem::path> files;
+            {
+                std::lock_guard l(this->m_fetchFilePathsMutex);
+                try {
+                    for (const auto& entry : std::filesystem::directory_iterator(this->m_queueDirectoryPath)) {
+                        files.push_back(entry.path());
+                    }
+                } catch (const std::exception& e) {
+                    spdlog::error("Error reading queue directory: {}", e.what());
+                }
+            }
+
+            for (const auto& filePath : files) {
+                auto fileName = filePath.filename().string();
+                spdlog::info("Processing file: {}", fileName);
+
+                size_t delimiterPos = fileName.find('-');
+                if (delimiterPos != std::string::npos) {
+                    std::string startedAt = fileName.substr(0, delimiterPos);
+                    std::string stoppedAt = fileName.substr(delimiterPos + 1);
+
+                    try {
+                        spdlog::info("Attempting to upload file...");
+                        if (SendRecordedFile(filePath.string(), stoull(startedAt), stoull(stoppedAt))) {
+                            spdlog::info("File uploaded, deleting: {}", filePath.string());
+                            std::filesystem::remove(filePath);
+                        } else {
+                            spdlog::warn("Failed to upload file: {}", filePath.string());
+                        }
+                    } catch (const std::exception& e) {
+                        spdlog::error("Exception during file upload: {}", e.what());
+                    }
+                }
+            }
+            std::this_thread::sleep_for(std::chrono::milliseconds(1000));
+        }
+    }
+
+
+
+    bool SendRecordedFile( const std::string& filepath, unsigned long long int startedAt, unsigned long long stoppedAt ) {
+        spdlog::info( "SendRecordedFile: {}", filepath );
+        httplib::Client client( this->m_configService->GetBaseUrl() );
+
+        std::ifstream ifs( filepath, std::ios::binary );
+        if( !ifs ) {
+            throw std::runtime_error( "Failed to open file" );
+        }
+
+        std::vector<char> buffer( ( std::istreambuf_iterator<char>( ifs ) ), ( std::istreambuf_iterator<char>() ) );
+        auto res = client.Post(
+            std::string( "/records/upload/" ),
+            httplib::MultipartFormDataItems{
+                { "file", std::string( buffer.begin(), buffer.end() ), "file.ogg", "audio/ogg" },
+                { "guid", this->m_configService->GetGuid() },
+                { "startedAt", std::to_string( startedAt ) },
+                { "stoppedAt", std::to_string( stoppedAt ) },
+            } );
+
+        if( res && res->status == 201 ) {
+            spdlog::info( "File uploaded successfully" );
+            return true;
+        }
+
+        spdlog::error( "Failed to upload file" );
+        return false;
+    }
+};
+
+}

src/Services/ConfigService.h

@@ -0,0 +1,76 @@
+#pragma once
+
+#include <string>
+#include <memory>
+#include <mutex>
+#include <nlohmann/json.hpp>
+#include <filesystem>
+#include <fstream>
+#include <sstream>
+
+namespace snoop {
+
+class Config {
+public:
+    std::string m_guid;
+    unsigned long long m_recordingDuration = 0;
+    std::string m_baseUrl;
+};
+
+NLOHMANN_DEFINE_TYPE_NON_INTRUSIVE_WITH_DEFAULT( Config, m_guid, m_recordingDuration, m_baseUrl )
+
+class ConfigService {
+    std::shared_ptr<Config> m_config;
+    std::string m_configFilePath;
+    std::mutex m_mutex;
+
+public:
+    explicit ConfigService( const std::string& configFilePath ) :
+        m_configFilePath( configFilePath ) {
+        if( !std::filesystem::exists( this->m_configFilePath ) ) {
+            throw std::runtime_error( std::string( "ConfigService: Config not found " ) + this->m_configFilePath );
+        }
+        std::string configFileContent = this->ReadFile( this->m_configFilePath );
+        nlohmann::json jsonConfig = nlohmann::json::parse( configFileContent );
+        this->m_config = std::make_shared<Config>( jsonConfig.get<Config>() );
+    }
+
+    [[nodiscard]] std::string GetGuid() const {
+        return this->m_config->m_guid;
+    }
+
+    [[nodiscard]] unsigned long long int GetRecordingDuration() const {
+        return this->m_config->m_recordingDuration;
+    }
+
+    void SetRecordingDuration( unsigned long long int recordingDuration ) {
+        this->m_config->m_recordingDuration = recordingDuration;
+        this->RewriteConfig();
+    }
+
+    [[nodiscard]] std::string GetBaseUrl() const {
+        return this->m_config->m_baseUrl;
+    }
+
+private:
+    std::string ReadFile( const std::string& path ) {
+        std::fstream f;
+        f.open( path, std::ios::in );
+        std::stringstream ss;
+        ss << f.rdbuf();
+        return ss.str();
+    }
+
+    void RewriteFile( const std::string& path, const std::string& fileContent ) {
+        std::fstream f;
+        f.open( path, std::ios::out );
+        f << fileContent;
+        f.close();
+    }
+
+    void RewriteConfig() {
+        this->RewriteFile( this->m_configFilePath, nlohmann::json( *this->m_config ).dump() );
+    }
+};
+
+}