rename the sequence receiver

This commit is contained in:
Sven Czarnian
2021-11-13 23:01:12 +01:00
parent 8b1292c95a
commit a3ef28ae39
4 changed files with 59 additions and 39 deletions

View File

@@ -1,6 +1,6 @@
/* /*
* @brief Defines the aircraft scheduling notifier module to receive plans from the backend * @brief Defines the aircraft scheduling notifier module to receive plans from the backend
* @file aman/com/AircraftScheduler.h * @file aman/com/BackendReceiver.h
* @author Sven Czarnian <devel@svcz.de> * @author Sven Czarnian <devel@svcz.de>
* @copyright Copyright 2021 Sven Czarnian * @copyright Copyright 2021 Sven Czarnian
* @license This project is published under the GNU General Public License v3 (GPLv3) * @license This project is published under the GNU General Public License v3 (GPLv3)
@@ -9,12 +9,13 @@
#pragma once #pragma once
#include <thread> #include <thread>
#include <memory>
#include <zmq.hpp> #include <zmq.hpp>
#include <aman/types/Communication.h> #include <aman/types/Communication.h>
#pragma warning(push, 0) #pragma warning(push, 0)
#include "protobuf/AircraftSchedule.pb.h" #include "protobuf/Communication.pb.h"
#pragma warning(pop) #pragma warning(pop)
namespace aman { namespace aman {
@@ -22,13 +23,15 @@ namespace aman {
* @brief Defines the aircraft scheduling notification class to receive scheduling sequences * @brief Defines the aircraft scheduling notification class to receive scheduling sequences
* @ingroup com * @ingroup com
*/ */
class AircraftScheduler { class BackendReceiver {
private: private:
std::unique_ptr<zmq::socket_t> m_socket; std::unique_ptr<zmq::socket_t> m_socket;
std::thread m_receiverThread; std::thread m_receiverThread;
std::atomic_bool m_stopReceiver; std::atomic_bool m_stopReceiver;
std::list<std::shared_ptr<aman::AircraftSequence>> m_sequences;
std::mutex m_sequencesLock;
AircraftScheduler() noexcept; BackendReceiver() noexcept;
template <typename T> template <typename T>
bool setSocketKey(const std::string& key, T entry) { bool setSocketKey(const std::string& key, T entry) {
@@ -40,13 +43,14 @@ namespace aman {
return false; return false;
} }
} }
void receiveSequence();
void run(); void run();
public: public:
AircraftScheduler(const AircraftScheduler&) = delete; BackendReceiver(const BackendReceiver&) = delete;
AircraftScheduler(AircraftScheduler&&) = delete; BackendReceiver(BackendReceiver&&) = delete;
AircraftScheduler& operator=(const AircraftScheduler&) = delete; BackendReceiver& operator=(const BackendReceiver&) = delete;
AircraftScheduler& operator=(AircraftScheduler&&) = delete; BackendReceiver& operator=(BackendReceiver&&) = delete;
/** /**
* @brief Initializes the aircraft scheduler * @brief Initializes the aircraft scheduler
@@ -64,14 +68,14 @@ namespace aman {
*/ */
bool initialized() const noexcept; bool initialized() const noexcept;
/** /**
* @brief Receives a new scheduling message * @brief Returns the current sequence of the receiver queue
* @return The new scheduling message or a nullptr * @return The sequence queue
*/ */
std::unique_ptr<aman::AircraftSchedule> receive(); std::shared_ptr<aman::AircraftSequence> receive();
/** /**
* @brief Returns the scheduling instance * @brief Returns the scheduling instance
* @return The system-wide instance * @return The system-wide instance
*/ */
static AircraftScheduler& instance() noexcept; static BackendReceiver& instance() noexcept;
}; };
} }

View File

@@ -32,8 +32,8 @@ SET(SOURCE_FILES
) )
SET(SOURCE_COM_FILES SET(SOURCE_COM_FILES
com/BackendReceiver.cpp
com/BackendNotification.cpp com/BackendNotification.cpp
com/AircraftScheduler.cpp
com/ZmqContext.cpp com/ZmqContext.cpp
com/ZmqContext.h com/ZmqContext.h
) )
@@ -51,8 +51,8 @@ SET(SOURCE_FILES_RES
) )
SET(INCLUDE_COM_FILES SET(INCLUDE_COM_FILES
${CMAKE_SOURCE_DIR}/include/aman/com/BackendReceiver.h
${CMAKE_SOURCE_DIR}/include/aman/com/BackendNotification.h ${CMAKE_SOURCE_DIR}/include/aman/com/BackendNotification.h
${CMAKE_SOURCE_DIR}/include/aman/com/AircraftScheduler.h
) )
SET(INCLUDE_CONFIG_FILES SET(INCLUDE_CONFIG_FILES

View File

@@ -15,7 +15,7 @@
#include <Shlwapi.h> #include <Shlwapi.h>
#include <Windows.h> #include <Windows.h>
#include <aman/com/AircraftScheduler.h> #include <aman/com/BackendReceiver.h>
#include <aman/com/BackendNotification.h> #include <aman/com/BackendNotification.h>
#include <aman/config/CommunicationFileFormat.h> #include <aman/config/CommunicationFileFormat.h>
#include <aman/config/IdentifierFileFormat.h> #include <aman/config/IdentifierFileFormat.h>
@@ -62,20 +62,20 @@ PlugIn::PlugIn() :
ZmqContext::instance().initialize(); ZmqContext::instance().initialize();
if (false == AircraftReporter::instance().initialize(this->m_configuration)) { if (false == BackendNotification::instance().initialize(this->m_configuration)) {
this->DisplayUserMessage(PLUGIN_NAME, "ERROR", "Unable to initialize the reporter-connection to the backend", true, true, true, true, true); this->DisplayUserMessage(PLUGIN_NAME, "ERROR", "Unable to initialize the reporter-connection to the backend", true, true, true, true, true);
return; return;
} }
if (false == AircraftScheduler::instance().initialize(this->m_configuration)) { if (false == BackendReceiver::instance().initialize(this->m_configuration)) {
this->DisplayUserMessage(PLUGIN_NAME, "ERROR", "Unable to initialize the scheduling-connection to the backend", true, true, true, true, true); this->DisplayUserMessage(PLUGIN_NAME, "ERROR", "Unable to initialize the scheduling-connection to the backend", true, true, true, true, true);
AircraftReporter::instance().deinitialize(); BackendNotification::instance().deinitialize();
return; return;
} }
} }
PlugIn::~PlugIn() noexcept { PlugIn::~PlugIn() noexcept {
AircraftScheduler::instance().deinitialize(); BackendReceiver::instance().deinitialize();
BackendNotification::instance().deinitialize(); BackendNotification::instance().deinitialize();
ZmqContext::instance().deinitialize(); ZmqContext::instance().deinitialize();
google::protobuf::ShutdownProtobufLibrary(); google::protobuf::ShutdownProtobufLibrary();

View File

@@ -2,26 +2,28 @@
* Author: * Author:
* Sven Czarnian <devel@svcz.de> * Sven Czarnian <devel@svcz.de>
* Brief: * Brief:
* Implements the scheduling notification * Implements the backend receiver
* Copyright: * Copyright:
* 2021 Sven Czarnian * 2021 Sven Czarnian
* License: * License:
* GNU General Public License v3 (GPLv3) * GNU General Public License v3 (GPLv3)
*/ */
#include <aman/com/AircraftScheduler.h> #include <aman/com/BackendReceiver.h>
#include "ZmqContext.h" #include "ZmqContext.h"
using namespace aman; using namespace aman;
using namespace std::chrono; using namespace std::chrono;
AircraftScheduler::AircraftScheduler() noexcept : BackendReceiver::BackendReceiver() noexcept :
m_socket(), m_socket(),
m_receiverThread(), m_receiverThread(),
m_stopReceiver(false) { } m_stopReceiver(false),
m_sequences(),
m_sequencesLock() { }
bool AircraftScheduler::initialize(const Communication& configuration) { bool BackendReceiver::initialize(const Communication& configuration) {
if (nullptr != this->m_socket) if (nullptr != this->m_socket)
return true; return true;
if (false == configuration.valid) if (false == configuration.valid)
@@ -47,12 +49,12 @@ bool AircraftScheduler::initialize(const Communication& configuration) {
} }
this->m_stopReceiver = false; this->m_stopReceiver = false;
this->m_receiverThread = std::thread(&AircraftScheduler::run, this); this->m_receiverThread = std::thread(&BackendReceiver::run, this);
return true; return true;
} }
bool AircraftScheduler::deinitialize() { bool BackendReceiver::deinitialize() {
if (nullptr == this->m_socket) if (nullptr == this->m_socket)
return true; return true;
@@ -64,42 +66,56 @@ bool AircraftScheduler::deinitialize() {
return true; return true;
} }
bool AircraftScheduler::initialized() const noexcept { bool BackendReceiver::initialized() const noexcept {
return nullptr != this->m_socket; return nullptr != this->m_socket;
} }
std::unique_ptr<aman::AircraftSchedule> AircraftScheduler::receive() { void BackendReceiver::receiveSequence() {
zmq::message_t message; zmq::message_t message;
if (nullptr == this->m_socket) if (nullptr == this->m_socket)
return std::unique_ptr<aman::AircraftSchedule>(); return;
try { try {
auto result = this->m_socket->recv(message, zmq::recv_flags::dontwait); auto result = this->m_socket->recv(message, zmq::recv_flags::dontwait);
if (false == result.has_value() || 0 == result.value()) if (false == result.has_value() || 0 == result.value())
return std::unique_ptr<aman::AircraftSchedule>(); return;
} }
catch (zmq::error_t&) { catch (zmq::error_t&) {
return std::unique_ptr<aman::AircraftSchedule>(); return;
} }
std::unique_ptr<aman::AircraftSchedule> retval = std::make_unique<aman::AircraftSchedule>(); std::unique_ptr<aman::AircraftSequence> retval = std::make_unique<aman::AircraftSequence>();
retval->ParseFromString(reinterpret_cast<const char*>(message.data())); retval->ParseFromString(reinterpret_cast<const char*>(message.data()));
return retval; std::lock_guard guard(this->m_sequencesLock);
this->m_sequences.push_back(std::move(retval));
} }
void AircraftScheduler::run() { void BackendReceiver::run() {
while (false == this->m_stopReceiver) { while (false == this->m_stopReceiver) {
auto message = this->receive(); auto message = this->receive();
if (nullptr == message) { if (nullptr == message) {
std::this_thread::sleep_for(500ms); std::this_thread::sleep_for(500ms);
continue; continue;
} }
this->receiveSequence();
} }
} }
AircraftScheduler& AircraftScheduler::instance() noexcept { std::shared_ptr<aman::AircraftSequence> BackendReceiver::receive() {
static AircraftScheduler __instance; std::lock_guard guard(this->m_sequencesLock);
if (0 != this->m_sequences.size()) {
auto retval = this->m_sequences.front();
this->m_sequences.pop_front();
return retval;
}
return nullptr;
}
BackendReceiver& BackendReceiver::instance() noexcept {
static BackendReceiver __instance;
return __instance; return __instance;
} }