From 218ee711180812fef4f01bbeceba3ee028c1dbd1 Mon Sep 17 00:00:00 2001 From: faraphel Date: Mon, 4 Nov 2024 23:40:40 +0100 Subject: [PATCH] reorganised the way the program share data (Context, Peer, RemotePeer) --- CMakeLists.txt | 5 +-- source/Context.hpp | 15 +++++++++ source/EventManager.cpp | 33 +++++++++---------- source/EventManager.hpp | 11 ++----- source/Peer.hpp | 15 --------- source/RemotePeer.hpp | 28 ++++++++++++++++ source/events/audio/AudioEvent.cpp | 14 ++++++-- source/events/audio/AudioEvent.hpp | 9 +++-- .../events/audio/AudioPacketsComparator.cpp | 2 +- .../events/audio/AudioPacketsComparator.hpp | 4 +-- source/events/base/BaseEvent.hpp | 8 ++++- source/events/info/InfoEvent.cpp | 21 ++++++++---- source/events/info/InfoEvent.hpp | 7 +++- source/events/pong/PongEvent.cpp | 7 +++- source/events/pong/PongEvent.hpp | 7 +++- source/events/search/SearchEvent.cpp | 21 +++++++----- source/events/search/SearchEvent.hpp | 7 +++- .../{AudioPacket.hpp => AudioPacketData.hpp} | 2 +- source/tasks/server/ServerTask.cpp | 9 +++-- source/tasks/server/ServerTask.hpp | 4 ++- source/tasks/undefined/UndefinedTask.cpp | 12 +++++-- source/utils/StatList.cpp | 6 ++++ source/utils/StatList.hpp | 6 ++++ source/utils/audio.cpp | 6 ++++ source/utils/audio.hpp | 8 ++++- 25 files changed, 190 insertions(+), 77 deletions(-) create mode 100644 source/Context.hpp delete mode 100644 source/Peer.hpp create mode 100644 source/RemotePeer.hpp rename source/packets/audio/{AudioPacket.hpp => AudioPacketData.hpp} (94%) diff --git a/CMakeLists.txt b/CMakeLists.txt index 409e24d..ca210f9 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -14,7 +14,7 @@ pkg_check_modules(OpenSSL REQUIRED openssl) add_executable(M2-PT-DRP source/main.cpp - source/packets/AudioPacket.hpp + source/packets/audio/AudioPacketData.hpp source/utils/audio.cpp source/utils/audio.hpp source/EventManager.cpp @@ -22,7 +22,7 @@ add_executable(M2-PT-DRP source/packets/base/GenericPacket.hpp source/events/types.hpp source/packets/base/GenericPacket.cpp - source/Peer.hpp + source/RemotePeer.hpp source/utils/StatList.cpp source/utils/StatList.hpp source/events/base/BaseEvent.hpp @@ -45,6 +45,7 @@ add_executable(M2-PT-DRP source/tasks/undefined/UndefinedTask.hpp source/tasks/client/ClientTask.cpp source/tasks/client/ClientTask.hpp + source/Context.hpp ) target_include_directories(M2-PT-DRP PRIVATE ${MPG123_INCLUDE_DIRS} diff --git a/source/Context.hpp b/source/Context.hpp new file mode 100644 index 0000000..05d3b13 --- /dev/null +++ b/source/Context.hpp @@ -0,0 +1,15 @@ +#pragma once + + +#include "RemotePeer.hpp" +#include "utils/StatList.hpp" + + +struct Context { + int socket = -1; + + drp::util::StatList> remotePeers {}; + std::shared_ptr server = nullptr; + + Peer me; +}; diff --git a/source/EventManager.cpp b/source/EventManager.cpp index 84f4c23..4fd4bae 100644 --- a/source/EventManager.cpp +++ b/source/EventManager.cpp @@ -2,7 +2,6 @@ #include #include -#include #include #include #include @@ -11,7 +10,6 @@ #include #include -#include "Peer.hpp" #include "events/types.hpp" #include "events/audio/AudioEvent.hpp" #include "events/info/InfoEvent.hpp" @@ -24,10 +22,6 @@ EventManager::EventManager() { - this->channel = 0; - this->server = nullptr; - this->status = drp::task::TaskType::UNDEFINED; - // register the different events type this->eventRegistry = { {drp::event::EventType::PONG, std::make_shared()}, @@ -51,13 +45,13 @@ EventManager::EventManager() { addressHints.ai_protocol = IPPROTO_UDP; // create the client socket - this->eventSocket = socket( + this->context.socket = socket( addressHints.ai_family, addressHints.ai_socktype, addressHints.ai_protocol ); - if (this->eventSocket < 0) + if (this->context.socket < 0) throw std::runtime_error("[Receiver] Could not create the socket: " + std::string(strerror(errno))); } @@ -95,7 +89,7 @@ void EventManager::loopSender() { // get the corresponding task class std::shared_ptr task; try { - task = this->taskRegistry.at(this->status); + task = this->taskRegistry.at(this->context.me.status); } catch (const std::out_of_range& exception) { std::cerr << "Unsupported status." << std::endl; continue; @@ -116,8 +110,8 @@ void EventManager::loopSender() { void EventManager::loopReceiver() { // prepare space for the sender address - sockaddr_storage senderAddress {}; - socklen_t senderAddressLength = sizeof(senderAddress); + sockaddr_storage fromAddress {}; + socklen_t fromAddressLength = sizeof(fromAddress); drp::packet::GenericPacket packet {}; drp::packet::GenericPacketContent packetContent {}; @@ -138,7 +132,7 @@ void EventManager::loopReceiver() { // bind the socket to the address if (bind( - this->eventSocket, + this->context.socket, senderInfo->ai_addr, senderInfo->ai_addrlen ) < 0) @@ -151,18 +145,18 @@ void EventManager::loopReceiver() { while (true) { // receive new data const ssize_t size = recvfrom( - this->eventSocket, + this->context.socket, &packet, sizeof(packet), 0, - reinterpret_cast(&senderAddress), - &senderAddressLength + reinterpret_cast(&fromAddress), + &fromAddressLength ); if (size == -1) throw std::runtime_error("[Receiver] Could not receive the packet: " + std::string(strerror(errno))); // if the packet channel is neither 0 (all) nor the current one, ignore it - if (packet.channel != 0 && packet.channel != this->channel) + if (packet.channel != 0 && packet.channel != this->context.me.channel) continue; // decrypt the packet @@ -178,6 +172,11 @@ void EventManager::loopReceiver() { } // ask the event class to handle the event - event->handle(packetContent); + event->handle( + this->context, + packetContent, + reinterpret_cast(&fromAddress), + fromAddressLength + ); } } diff --git a/source/EventManager.hpp b/source/EventManager.hpp index df65d95..3a17415 100644 --- a/source/EventManager.hpp +++ b/source/EventManager.hpp @@ -5,7 +5,8 @@ #include #include -#include "Peer.hpp" +#include "Context.hpp" +#include "RemotePeer.hpp" #include "events/types.hpp" #include "events/base/BaseEvent.hpp" #include "tasks/types.hpp" @@ -28,11 +29,5 @@ private: std::map> eventRegistry; /// hold the event to call depending on the event type std::map> taskRegistry; /// hold the task to call depending on the server status - StatList> peers; /// list of peers found - std::shared_ptr server; /// the peer used as a server - - int eventSocket; /// the socket used to communicate - drp::task::TaskType status; /// our current status - - std::uint8_t channel; /// the packet channel currently used + Context context; /// context used between the events types }; diff --git a/source/Peer.hpp b/source/Peer.hpp deleted file mode 100644 index 04254e5..0000000 --- a/source/Peer.hpp +++ /dev/null @@ -1,15 +0,0 @@ -#pragma once - -#include - -#include "tasks/types.hpp" - - -struct Peer { - // communication - sockaddr address; - socklen_t addressLength; - - // information - drp::task::TaskType status; -}; diff --git a/source/RemotePeer.hpp b/source/RemotePeer.hpp new file mode 100644 index 0000000..977ad06 --- /dev/null +++ b/source/RemotePeer.hpp @@ -0,0 +1,28 @@ +#pragma once + +#include + +#include "tasks/types.hpp" + + +/** + * Contains common information about a certain peer. + */ +struct Peer { + bool serverEnabled = false; + drp::task::TaskType status = drp::task::TaskType::UNDEFINED; + uint8_t channel = 0; +}; + + +/** + * Contains information about a distant peer. + */ +struct RemotePeer { + // communication + sockaddr address {}; + socklen_t addressLength = 0; + + // information + Peer information; +}; diff --git a/source/events/audio/AudioEvent.cpp b/source/events/audio/AudioEvent.cpp index 3e62f42..5f97dbe 100644 --- a/source/events/audio/AudioEvent.cpp +++ b/source/events/audio/AudioEvent.cpp @@ -1,5 +1,6 @@ #include "AudioEvent.hpp" +#include #include #include @@ -28,8 +29,17 @@ AudioEvent::~AudioEvent() { } -void AudioEvent::handle(const packet::GenericPacketContent& content) { - this->audioQueue.push(static_cast(content)); +void AudioEvent::handle( + Context& context, + const packet::GenericPacketContent& content, + sockaddr* fromAddress, + socklen_t fromAddressLength +) { + // get the audio data in the content + packet::AudioPacketData audioData; + std::memcpy(&audioData, content.data.data(), content.data.size()); + // save it in the audio queue + this->audioQueue.push(audioData); // notify that a new audio chunk is available this->audioCondition.notify_one(); } diff --git a/source/events/audio/AudioEvent.hpp b/source/events/audio/AudioEvent.hpp index e0261bb..369a996 100644 --- a/source/events/audio/AudioEvent.hpp +++ b/source/events/audio/AudioEvent.hpp @@ -20,7 +20,12 @@ public: void updateAudioStream(int channels, std::uint32_t sampleFormat, double sampleRate); void loopPlay(); - void handle(const packet::GenericPacketContent& content) override; + void handle( + Context& context, + const packet::GenericPacketContent& content, + sockaddr* fromAddress, + socklen_t fromAddressLength + ) override; private: std::thread playerThread; @@ -29,7 +34,7 @@ private: int streamChannels; std::uint32_t streamSampleFormat; double streamRate; - std::priority_queue, AudioPacketsComparator> audioQueue; + std::priority_queue, AudioPacketsComparator> audioQueue; std::mutex audioMutex; std::unique_lock audioLock; diff --git a/source/events/audio/AudioPacketsComparator.cpp b/source/events/audio/AudioPacketsComparator.cpp index b5ba2bb..c3fbe6c 100644 --- a/source/events/audio/AudioPacketsComparator.cpp +++ b/source/events/audio/AudioPacketsComparator.cpp @@ -4,7 +4,7 @@ namespace drp::event { -bool AudioPacketsComparator::operator()(const packet::AudioPacket& a, const packet::AudioPacket& b) const { +bool AudioPacketsComparator::operator()(const packet::AudioPacketData& a, const packet::AudioPacketData& b) const { return a.timePlay > b.timePlay; } diff --git a/source/events/audio/AudioPacketsComparator.hpp b/source/events/audio/AudioPacketsComparator.hpp index 6e85324..b1118fe 100644 --- a/source/events/audio/AudioPacketsComparator.hpp +++ b/source/events/audio/AudioPacketsComparator.hpp @@ -1,13 +1,13 @@ #pragma once -#include "../../packets/audio/AudioPacket.hpp" +#include "../../packets/audio/AudioPacketData.hpp" namespace drp::event { struct AudioPacketsComparator { - bool operator() (const packet::AudioPacket& a, const packet::AudioPacket& b) const; + bool operator() (const packet::AudioPacketData& a, const packet::AudioPacketData& b) const; }; diff --git a/source/events/base/BaseEvent.hpp b/source/events/base/BaseEvent.hpp index 179fe87..1611e06 100644 --- a/source/events/base/BaseEvent.hpp +++ b/source/events/base/BaseEvent.hpp @@ -1,6 +1,7 @@ #pragma once #include "../../packets/base/GenericPacket.hpp" +#include "../../Context.hpp" namespace drp::event { @@ -9,7 +10,12 @@ namespace drp::event { class BaseEvent { public: virtual ~BaseEvent() = default; - virtual void handle(const packet::GenericPacketContent& content) = 0; + virtual void handle( + Context& context, + const packet::GenericPacketContent& content, + sockaddr* fromAddress, + socklen_t fromAddressLength + ) = 0; }; diff --git a/source/events/info/InfoEvent.cpp b/source/events/info/InfoEvent.cpp index d4d52ee..6d839ed 100644 --- a/source/events/info/InfoEvent.cpp +++ b/source/events/info/InfoEvent.cpp @@ -4,17 +4,26 @@ #include "../../tasks/types.hpp" + namespace drp::event { -void InfoEvent::handle(const packet::GenericPacketContent& content) { - Peer peer {}; - peer.address = *reinterpret_cast(&senderAddress); - peer.addressLength = senderAddressLength; - peer.status = task::TaskType::UNDEFINED; +void InfoEvent::handle( + Context& context, + const packet::GenericPacketContent& content, + sockaddr* fromAddress, + const socklen_t fromAddressLength +) { + const auto remotePeer = std::make_shared(); + + remotePeer->address = *reinterpret_cast(&fromAddress); + remotePeer->addressLength = fromAddressLength; + + // TODO(Faraphel): convert the memory from content to Peer. + remotePeer->information = content; // save it in the peers list - this->peers.push_back(peer); + context.remotePeers.push_back(remotePeer); } diff --git a/source/events/info/InfoEvent.hpp b/source/events/info/InfoEvent.hpp index a4eb56f..cd5b6a0 100644 --- a/source/events/info/InfoEvent.hpp +++ b/source/events/info/InfoEvent.hpp @@ -8,7 +8,12 @@ namespace drp::event { class InfoEvent : public BaseEvent { public: - void handle(const packet::GenericPacketContent& content) override; + void handle( + Context& context, + const packet::GenericPacketContent& content, + sockaddr* fromAddress, + socklen_t fromAddressLength + ) override; }; diff --git a/source/events/pong/PongEvent.cpp b/source/events/pong/PongEvent.cpp index 0c8b5d3..751ab6f 100644 --- a/source/events/pong/PongEvent.cpp +++ b/source/events/pong/PongEvent.cpp @@ -6,7 +6,12 @@ namespace drp::event { -void PongEvent::handle(const packet::GenericPacketContent& content) { +void PongEvent::handle( + Context& context, + const packet::GenericPacketContent& content, + sockaddr* fromAddress, + socklen_t fromAddressLength +) { std::cout << "[Receiver] Pong." << std::endl; } diff --git a/source/events/pong/PongEvent.hpp b/source/events/pong/PongEvent.hpp index 8c6ec18..2a629e9 100644 --- a/source/events/pong/PongEvent.hpp +++ b/source/events/pong/PongEvent.hpp @@ -8,7 +8,12 @@ namespace drp::event { class PongEvent : public BaseEvent { public: - void handle(const packet::GenericPacketContent& content) override; + void handle( + Context& context, + const packet::GenericPacketContent& content, + sockaddr* fromAddress, + socklen_t fromAddressLength + ) override; }; diff --git a/source/events/search/SearchEvent.cpp b/source/events/search/SearchEvent.cpp index 4787a98..1a890ce 100644 --- a/source/events/search/SearchEvent.cpp +++ b/source/events/search/SearchEvent.cpp @@ -13,27 +13,32 @@ namespace drp { -void event::SearchEvent::handle(const packet::GenericPacketContent& content) { +void event::SearchEvent::handle( + Context& context, + const packet::GenericPacketContent& content, + sockaddr* fromAddress, + const socklen_t fromAddressLength +) { packet::GenericPacket packet {}; packet::GenericPacketContent packetContent {}; - Information information {}; - information.status = task::TaskType::UNDEFINED; - + // create the packet header (available to read for everyone) packet.channel = 0; packet.securityMode = static_cast(packet::SecurityMode::PLAIN); + // insert our information into the packet packetContent.eventType = static_cast(EventType::INFO); - std::memcpy(&packetContent.data, &information, sizeof(Information)); + std::memcpy(&packetContent.data, &context.me, sizeof(context.me)); packet.setContent(packetContent); + // broadcast our information if (sendto( - clientSocket, + context.socket, &packet, sizeof(packet), 0, - serverAddress, - serverAddressLength + fromAddress, + fromAddressLength ) == -1) std::cerr << "[Receiver] Could not send information: " << strerror(errno) << std::endl; diff --git a/source/events/search/SearchEvent.hpp b/source/events/search/SearchEvent.hpp index cf6b1ba..fadd7c4 100644 --- a/source/events/search/SearchEvent.hpp +++ b/source/events/search/SearchEvent.hpp @@ -7,7 +7,12 @@ namespace drp::event { class SearchEvent : public BaseEvent { public: - void handle(const packet::GenericPacketContent& content) override; + void handle( + Context& context, + const packet::GenericPacketContent& content, + sockaddr* fromAddress, + socklen_t fromAddressLength + ) override; }; diff --git a/source/packets/audio/AudioPacket.hpp b/source/packets/audio/AudioPacketData.hpp similarity index 94% rename from source/packets/audio/AudioPacket.hpp rename to source/packets/audio/AudioPacketData.hpp index 90f1faf..b354d8a 100644 --- a/source/packets/audio/AudioPacket.hpp +++ b/source/packets/audio/AudioPacketData.hpp @@ -7,7 +7,7 @@ namespace drp::packet { -struct AudioPacket { +struct AudioPacketData { // scheduling // TODO(Faraphel): use a more "fixed" size format ? std::chrono::time_point timePlay; diff --git a/source/tasks/server/ServerTask.cpp b/source/tasks/server/ServerTask.cpp index 53ce685..768c00b 100644 --- a/source/tasks/server/ServerTask.cpp +++ b/source/tasks/server/ServerTask.cpp @@ -10,6 +10,9 @@ #include #include +#include "../../packets/audio/AudioPacketData.hpp" +#include "../../utils/audio.hpp" + namespace drp::task { @@ -78,7 +81,7 @@ void ServerTask::loop() const { throw std::runtime_error("[Server] Could not create the socket: " + std::string(gai_strerror(errno))); // read the file - AudioPacket audioPacket; + packet::AudioPacketData audioPacket; std::size_t done; while (mpg123_read( @@ -95,7 +98,7 @@ void ServerTask::loop() const { // set the audio settings audioPacket.channels = this->channels; - audioPacket.sampleFormat = encoding_mpg123_to_PulseAudio(this->encoding); + audioPacket.sampleFormat = util::encoding_mpg123_to_PulseAudio(this->encoding); audioPacket.sampleRate = this->sampleRate; // set the size of the content @@ -129,4 +132,4 @@ void ServerTask::loop() const { } -} \ No newline at end of file +} diff --git a/source/tasks/server/ServerTask.hpp b/source/tasks/server/ServerTask.hpp index cd6e0f2..a2777b6 100644 --- a/source/tasks/server/ServerTask.hpp +++ b/source/tasks/server/ServerTask.hpp @@ -1,6 +1,8 @@ #pragma once #include +#include "../base/BaseTask.hpp" + namespace drp::task { @@ -25,4 +27,4 @@ private: }; -} \ No newline at end of file +} diff --git a/source/tasks/undefined/UndefinedTask.cpp b/source/tasks/undefined/UndefinedTask.cpp index fef9826..c4caa62 100644 --- a/source/tasks/undefined/UndefinedTask.cpp +++ b/source/tasks/undefined/UndefinedTask.cpp @@ -26,13 +26,18 @@ void UndefinedTask::handle() { ) { // verify if there are peers if (this->peers.empty()) { - // if we are alone, become the server - // TODO(Faraphel): only apply if we are capable of emitting ! + // if we are alone in the network + + // check if we are capable of being a server + if (!canEmit()) + return; + + // set ourselves as the server this->status = TaskType::SERVER; return; } - // search for a server in the peers + // search for a server among the peers const auto& server = std::find_if( this->peers.begin(), this->peers.end(), @@ -51,6 +56,7 @@ void UndefinedTask::handle() { } // prepare a search message + packet::GenericPacket packet {}; packet.channel = 0; packet.securityMode = static_cast(packet::SecurityMode::PLAIN); packet._content.eventType = static_cast(event::EventType::SEARCH); diff --git a/source/utils/StatList.cpp b/source/utils/StatList.cpp index b014cfd..9f0e302 100644 --- a/source/utils/StatList.cpp +++ b/source/utils/StatList.cpp @@ -1,6 +1,9 @@ #include "StatList.hpp" +namespace drp::util { + + template StatList::StatList() { this->updateModificationTime(); @@ -28,3 +31,6 @@ template void StatList::updateModificationTime() { this->modificationTime = std::chrono::high_resolution_clock::now(); } + + +} \ No newline at end of file diff --git a/source/utils/StatList.hpp b/source/utils/StatList.hpp index 3565e89..b21b0a8 100644 --- a/source/utils/StatList.hpp +++ b/source/utils/StatList.hpp @@ -4,6 +4,9 @@ #include +namespace drp::util { + + /** * A simple list with additional stat data, such as the date of the latest modification * @tparam T the type of elements in the list @@ -22,3 +25,6 @@ private: void updateModificationTime(); std::chrono::time_point modificationTime; }; + + +} \ No newline at end of file diff --git a/source/utils/audio.cpp b/source/utils/audio.cpp index bfee1e0..75e4053 100644 --- a/source/utils/audio.cpp +++ b/source/utils/audio.cpp @@ -6,6 +6,9 @@ #include +namespace drp::util { + + std::uint32_t encoding_mpg123_to_PulseAudio(const int encoding_mpg123) { switch (encoding_mpg123) { case MPG123_ENC_UNSIGNED_8: @@ -25,3 +28,6 @@ std::uint32_t encoding_mpg123_to_PulseAudio(const int encoding_mpg123) { throw std::runtime_error("Invalid encoding value."); } } + + +} \ No newline at end of file diff --git a/source/utils/audio.hpp b/source/utils/audio.hpp index 3e6526d..a5e7ac4 100644 --- a/source/utils/audio.hpp +++ b/source/utils/audio.hpp @@ -3,4 +3,10 @@ #include -std::uint32_t encoding_mpg123_to_PulseAudio(int encoding_mpg123); \ No newline at end of file +namespace drp::util { + + +std::uint32_t encoding_mpg123_to_PulseAudio(int encoding_mpg123); + + +} \ No newline at end of file